minor update on sarama config (#461)

pingcap · Feb 26, 2019 · 9ee817f · 9ee817f
1 parent 50f5b19
commit 9ee817f
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 44 deletions.
diff --git a/drainer/util.go b/drainer/util.go
@@ -67,7 +67,8 @@ func GenCheckPointCfg(cfg *Config, id uint64) *checkpoint.Config {
 
 func initializeSaramaGlobalConfig() {
 	sarama.MaxResponseSize = int32(maxMsgSize)
-	sarama.MaxRequestSize = int32(maxMsgSize)
+	// add 1 to avoid confused log: Producer.MaxMessageBytes must be smaller than MaxRequestSize; it will be ignored
+	sarama.MaxRequestSize = int32(maxMsgSize) + 1
 }
 
 func getDDLJob(tiStore kv.Storage, id int64) (*model.Job, error) {

diff --git a/pkg/util/kafka.go b/pkg/util/kafka.go
@@ -26,42 +26,6 @@ func initMetrics() {
 	exp.Exp(metricRegistry)
 }
 
-// CreateKafkaProducer create a sync producer
-func CreateKafkaProducer(config *sarama.Config, addr []string, kafkaVersion string, maxMsgSize int, metricsPrefix string) (sarama.SyncProducer, error) {
-	var (
-		client sarama.SyncProducer
-		err    error
-	)
-
-	// initial kafka client to use manual partitioner
-	if config == nil {
-		config = sarama.NewConfig()
-	}
-	config.Producer.Partitioner = sarama.NewManualPartitioner
-	config.Producer.MaxMessageBytes = maxMsgSize
-	config.Producer.Return.Successes = true
-	config.Producer.RequiredAcks = sarama.WaitForAll
-	version, err := sarama.ParseKafkaVersion(kafkaVersion)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-	config.Version = version
-	config.MetricRegistry = metrics.NewPrefixedChildRegistry(GetParentMetricsRegistry(), metricsPrefix)
-
-	log.Infof("kafka producer version %v", version)
-	for i := 0; i < maxRetry; i++ {
-		client, err = sarama.NewSyncProducer(addr, config)
-		if err != nil {
-			log.Errorf("create kafka client error %v", err)
-			time.Sleep(retryInterval)
-			continue
-		}
-		return client, nil
-	}
-
-	return nil, errors.Trace(err)
-}
-
 // GetParentMetricsRegistry get the metrics registry and expose the metrics while /debug/metrics
 func GetParentMetricsRegistry() metrics.Registry {
 	metricRegistryOnce.Do(initMetrics)
@@ -77,25 +41,22 @@ func NewSaramaConfig(kafkaVersion string, metricsPrefix string) (*sarama.Config,
 		return nil, errors.Trace(err)
 	}
 
+	config.ClientID = "tidb_binlog"
 	config.Version = version
+	log.Debugf("kafka consumer version %v", version)
 	config.MetricRegistry = metrics.NewPrefixedChildRegistry(GetParentMetricsRegistry(), metricsPrefix)
 
 	return config, nil
 }
 
 // CreateKafkaConsumer creates a kafka consumer
 func CreateKafkaConsumer(kafkaAddrs []string, kafkaVersion string) (sarama.Consumer, error) {
-	kafkaCfg := sarama.NewConfig()
-	kafkaCfg.Consumer.Return.Errors = true
-	version, err := sarama.ParseKafkaVersion(kafkaVersion)
+	kafkaCfg, err := NewSaramaConfig(kafkaVersion, "drainer.")
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	kafkaCfg.Version = version
-	log.Infof("kafka consumer version %v", version)
 
-	registry := GetParentMetricsRegistry()
-	kafkaCfg.MetricRegistry = metrics.NewPrefixedChildRegistry(registry, "drainer.")
+	kafkaCfg.Consumer.Return.Errors = true
 
 	return sarama.NewConsumer(kafkaAddrs, kafkaCfg)
 }