drainer: add KafkaClientID in DBConfig (#902)

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

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

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:"-"`

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