fixing test

idk/kafka_sasl/cmd.go

@@ -5,7 +5,6 @@ import (
 
 	confluent "github.com/confluentinc/confluent-kafka-go/kafka"
 	"github.com/featurebasedb/featurebase/v3/idk"
-	"github.com/featurebasedb/featurebase/v3/idk/common"
 	"github.com/pkg/errors"
 )
 
@@ -21,17 +20,6 @@ type Main struct {
 	ConfigMap            *confluent.ConfigMap `flag:"-"`
 }
 
-func (m *Main) CopyIn(config idk.ConfluentCommand) {
-	m.KafkaSaslUsername = config.KafkaSaslUsername
-	m.KafkaSaslPassword = config.KafkaSaslPassword
-	m.KafkaSaslMechanism = config.KafkaSaslMechanism
-	m.KafkaSecurityProtocol = config.KafkaSecurityProtocol
-	m.KafkaSslKeyPassword = config.KafkaSslKeyPassword
-	m.KafkaSslCaLocation = config.KafkaSslCaLocation
-	m.KafkaSslCertificateLocation = config.KafkaSslCertificateLocation
-	m.KafkaSslKeyLocation = config.KafkaSslKeyLocation
-}
-
 func NewMain() (*Main, error) {
 	var err error
 	m := Main{}
@@ -49,20 +37,26 @@ func NewMain() (*Main, error) {
 	}
 	m.NewSource = func() (idk.Source, error) {
 		source := NewSource()
+		source.KafkaBootstrapServers = m.KafkaBootstrapServers
+		source.SchemaRegistryURL = m.SchemaRegistryURL
 		source.Group = m.Group
 		source.Topics = m.Topics
 		source.Log = m.Main.Log()
 		source.Timeout = m.Timeout
+		source.KafkaSocketTimeoutMs = int(m.Timeout / time.Millisecond)
 		source.SkipOld = m.SkipOld
+		source.ConfluentCommand = m.ConfluentCommand
+		source.SchemaRegistryUsername = m.SchemaRegistryUsername
+		source.SchemaRegistryPassword = m.SchemaRegistryPassword
+		source.Verbose = m.Verbose
+		source.KafkaMaxPollInterval = m.KafkaMaxPollInterval
+		source.KafkaSessionTimeout = m.KafkaSessionTimeout
+		source.KafkaGroupInstanceId = m.KafkaGroupInstanceId
+		source.KafkaDebug = m.KafkaDebug
+		source.KafkaSocketKeepaliveEnable = m.KafkaSocketKeepaliveEnable
 		source.Header = m.Header
 		source.AllowMissingFields = m.AllowMissingFields
-		source.KafkaBootstrapServers = m.KafkaBootstrapServers
-		cfg, err := common.SetupConfluent(&m.ConfluentCommand)
-		if err != nil {
-			return nil, err
-		}
 
-		source.ConfigMap = cfg
 		err = source.Open()
 		if err != nil {
 			return nil, errors.Wrap(err, "opening source")