kafka: add User and Password #968

actiontech · Dec 12, 2022 · 768c046 · 768c046
1 parent ab25514
commit 768c046
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 3 deletions.
diff --git a/driver/common/taskconfig.go b/driver/common/taskconfig.go
@@ -142,6 +142,8 @@ type KafkaConfig struct {
 	Topic               string
 	Converter           string
 	TimeZone            string
+	User                string
+	Password            string
 	MessageGroupMaxSize uint64
 	MessageGroupTimeout uint64
 

diff --git a/driver/driver.go b/driver/driver.go
@@ -82,7 +82,7 @@ var (
 			hclspec.NewLiteral(`""`)),
 		"memory":          hclspec.NewAttr("memory", "string", false),
 		"big_tx_max_jobs": hclspec.NewAttr("big_tx_max_jobs", "number", false),
-		"log_file":        hclspec.NewDefault(hclspec.NewAttr("log_file", "string", false),
+		"log_file": hclspec.NewDefault(hclspec.NewAttr("log_file", "string", false),
 			hclspec.NewLiteral(`"/var/log/dtle"`)),
 	})
 
@@ -164,6 +164,8 @@ var (
 			"TopicWithSchemaTable": hclspec.NewDefault(hclspec.NewAttr("TopicWithSchemaTable", "bool", false),
 				hclspec.NewLiteral(`true`)),
 			"SchemaChangeTopic": hclspec.NewAttr("SchemaChangeTopic", "string", false),
+			"User":              hclspec.NewAttr("User", "string", false),
+			"Password":          hclspec.NewAttr("SchemaChangeTopic", "string", false),
 		})),
 		// Since each job has its own history, this should be smaller than MySQL default (25000).
 		"DependencyHistorySize": hclspec.NewDefault(hclspec.NewAttr("DependencyHistorySize", "number", false),
@@ -229,8 +231,8 @@ type Driver struct {
 	// logger will log to the Nomad agent
 	logger g.LoggerType
 
-	stand     *stand.StanServer
-	apiServer *httprouter.Router
+	stand            *stand.StanServer
+	apiServer        *httprouter.Router
 	setupApiServerFn func(logger g.LoggerType, driverConfig *DriverConfig) error
 
 	config *DriverConfig

diff --git a/driver/kafka/kafka2.go b/driver/kafka/kafka2.go
@@ -71,6 +71,8 @@ func NewKafkaManager(kcfg *common.KafkaConfig) (*KafkaManager, error) {
 	}
 	config := sarama.NewConfig()
 	config.Producer.Return.Successes = true
+	config.Net.SASL.User = kcfg.User
+	config.Net.SASL.Password = kcfg.Password
 
 	k.producer, err = sarama.NewSyncProducer(kcfg.Brokers, config)
 	if err != nil {