increase rate of metadata fetching in tests

acceptance_test.go

@@ -17,6 +17,7 @@ package kafka
 import (
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/conduitio/conduit-connector-kafka/common"
 	"github.com/conduitio/conduit-connector-kafka/source"
@@ -53,6 +54,9 @@ func TestAcceptance(t *testing.T) {
 					"TestSource_Configure_RequiredParams",
 					"TestDestination_Configure_RequiredParams",
 				},
+
+				WriteTimeout: time.Second * 10,
+				ReadTimeout:  time.Second * 10,
 			},
 		},
 	})

common/config.go

@@ -41,6 +41,8 @@ type Config struct {
 
 	ConfigSASL
 	ConfigTLS
+
+	franzClientOpts []kgo.Opt
 }
 
 // Validate executes manual validations beyond what is defined in struct tags.

common/franz.go

@@ -21,12 +21,20 @@ import (
 	"github.com/twmb/franz-go/pkg/kgo"
 )
 
+// WithFranzClientOpts lets you specify custom kafka client options (meant for
+// test purposes).
+func (c Config) WithFranzClientOpts(opts ...kgo.Opt) Config {
+	c.franzClientOpts = append(c.franzClientOpts, opts...)
+	return c
+}
+
 // FranzClientOpts returns the kafka client options derived from the common config.
 func (c Config) FranzClientOpts(logger *zerolog.Logger) []kgo.Opt {
 	opts := []kgo.Opt{
 		kgo.SeedBrokers(c.Servers...),
 		kgo.ClientID(c.ClientID),
 	}
+	opts = append(opts, c.franzClientOpts...)
 	if logger.GetLevel() != zerolog.Disabled {
 		opts = append(opts, kgo.WithLogger(franzLogger{logger: logger}))
 	}

destination/franz_integration_test.go

@@ -29,6 +29,7 @@ func TestFranzProducer_Produce(t *testing.T) {
 	ctx := context.Background()
 
 	cfg := test.ParseConfigMap[Config](t, test.DestinationConfigMap(t))
+	cfg.Config = test.ConfigWithIntegrationTestOptions(cfg.Config)
 
 	p, err := NewFranzProducer(ctx, cfg)
 	is.NoErr(err)

test/util.go

@@ -100,6 +100,7 @@ func Consume(t T, cfg common.Config, limit int) []*kgo.Record {
 	cl, err := kgo.NewClient(
 		kgo.SeedBrokers(cfg.Servers...),
 		kgo.ConsumeTopics(cfg.Topic),
+		kgo.MetadataMinAge(time.Millisecond*100),
 	)
 	is.NoErr(err)
 	defer cl.Close()
@@ -129,6 +130,7 @@ func Produce(t T, cfg common.Config, records []*kgo.Record, timeoutOpt ...time.D
 	cl, err := kgo.NewClient(
 		kgo.SeedBrokers(cfg.Servers...),
 		kgo.DefaultProduceTopic(cfg.Topic),
+		kgo.MetadataMinAge(time.Millisecond*100),
 	)
 	is.NoErr(err)
 	defer cl.Close()
@@ -173,6 +175,7 @@ func CreateTopic(t T, cfg common.Config) {
 
 	cl, err := kgo.NewClient(
 		kgo.SeedBrokers(cfg.Servers...),
+		kgo.MetadataMinAge(time.Millisecond*100),
 	)
 	is.NoErr(err)
 
@@ -220,3 +223,11 @@ func Certificates(t T) (clientCert, clientKey, caCert string) {
 	caCert = readFile("server.cer.pem")
 	return
 }
+
+func ConfigWithIntegrationTestOptions(cfg common.Config) common.Config {
+	return cfg.WithFranzClientOpts(
+		// by default metadata is fetched every 5 seconds, for integration tests
+		// we set this to a lower value so the tests finish faster
+		kgo.MetadataMinAge(time.Millisecond * 100),
+	)
+}