drainer: add KafkaClientID in DBConfig (#902) (#929)

Improve configuration for kafka syncer: Add kafka-client-id in drainer.toml to config kafka client.id property. Co-authored-by: dixingxing <[email protected]>
pingcap · Mar 6, 2020 · e4f2ba2 · e4f2ba2
1 parent 56ced1e
commit e4f2ba2
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 0 deletions.
diff --git a/cmd/drainer/drainer.toml b/cmd/drainer/drainer.toml
@@ -137,6 +137,7 @@ port = 3306
 # kafka-addrs = "127.0.0.1:9092"
 # kafka-version = "0.8.2.0"
 # kafka-max-messages = 1024
+# kafka-client-id = "tidb_binlog"
 #
 #
 # the topic name drainer will push msg, the default name is <cluster-id>_obinlog

diff --git a/drainer/sync/kafka.go b/drainer/sync/kafka.go
@@ -77,6 +77,10 @@ func NewKafka(cfg *DBConfig, tableInfoGetter translator.TableInfoGetter) (*Kafka
 		return nil, errors.Trace(err)
 	}
 
+	if len(cfg.KafkaClientID) > 0 {
+		config.ClientID = cfg.KafkaClientID
+	}
+
 	config.Producer.Flush.MaxMessages = cfg.KafkaMaxMessages
 
 	// maintain minimal set that has been necessary so far

diff --git a/drainer/sync/util.go b/drainer/sync/util.go
@@ -34,6 +34,7 @@ type DBConfig struct {
 	KafkaAddrs       string `toml:"kafka-addrs" json:"kafka-addrs"`
 	KafkaVersion     string `toml:"kafka-version" json:"kafka-version"`
 	KafkaMaxMessages int    `toml:"kafka-max-messages" json:"kafka-max-messages"`
+	KafkaClientID    string `toml:"kafka-client-id" json:"kafka-client-id"`
 	TopicName        string `toml:"topic-name" json:"topic-name"`
 	// get it from pd
 	ClusterID uint64 `toml:"-" json:"-"`

diff --git a/tests/binlog/drainer.toml b/tests/binlog/drainer.toml
@@ -75,6 +75,7 @@ port = 3306
 # kafka-addrs = "127.0.0.1:9092"
 # kafka-version = "0.8.2.0"
 # kafka-max-messages = 1024
+# kafka-client-id = "tidb_binlog"
 #
 #
 # the topic name drainer will push msg, the default name is <cluster-id>_obinlog