Merge pull request #1214 from PapaPiya/fix_kafka_sender

[PDR-17200][feat(kafka)]: 增加kafka sender version配置

Author: lvheyang

sender/config/config.go

@@ -986,6 +986,14 @@ var ModeKeyOptions = map[string][]Option{
 			Description:  "SASL密码(sasl_password)",
 			ToolTip:      "使用SASL协议进行鉴权的密码",
 		},
+		{
+			KeyName:      KeyKafkaVersion,
+			ChooseOnly:   false,
+			Default:      "",
+			DefaultNoUse: false,
+			Description:  "kafka版本(kafka_version)",
+			Advance:      true,
+		},
 		{
 			KeyName:      KeyKafkaRetryMax,
 			ChooseOnly:   false,

sender/config/models.go

@@ -222,6 +222,7 @@ const (
 	KeyKafkaCompression               = "kafka_compression"      //压缩模式,有none, gzip, snappy
 	KeyKafkaTimeout                   = "kafka_timeout"          //连接超时时间
 	KeyKafkaKeepAlive                 = "kafka_keep_alive"       //保持连接时长
+	KeyKafkaVersion                   = "kafka_version"          //版本
 	KeyMaxMessageBytes                = "max_message_bytes"      //每条消息最大字节数
 	KeyGZIPCompressionLevel           = "gzip_compression_level" //GZIP压缩日志的策略
 	KeyGZIPCompressionNo              = "仅打包不压缩"

sender/kafka/kafka.go

@@ -9,7 +9,7 @@ import (
 	"time"
 
 	"github.com/Shopify/sarama"
-	"github.com/json-iterator/go"
+	jsoniter "github.com/json-iterator/go"
 	"github.com/rcrowley/go-metrics"
 
 	"github.com/qiniu/log"
@@ -86,6 +86,7 @@ func NewSender(conf conf.MapConf) (kafkaSender sender.Sender, err error) {
 	keepAlive, _ := conf.GetStringOr(KeyKafkaKeepAlive, "0")
 	maxMessageBytes, _ := conf.GetIntOr(KeyMaxMessageBytes, 4*1024*1024)
 	gzipCompressionLevel, _ := conf.GetStringOr(KeyGZIPCompressionLevel, KeyGZIPCompressionDefault)
+	version, _ := conf.GetStringOr(KeyKafkaVersion, "")
 
 	name, _ := conf.GetStringOr(KeyName, fmt.Sprintf("kafkaSender:(kafkaUrl:%s,topic:%s)", hosts, topic))
 
@@ -108,8 +109,10 @@ func NewSender(conf conf.MapConf) (kafkaSender sender.Sender, err error) {
 	if !ok {
 		return nil, fmt.Errorf("unknown compression mode: '%v'", compression)
 	}
-	if compressionMode == sarama.CompressionLZ4 {
-		cfg.Version = sarama.V0_10_0_0
+	if version != "" {
+		if cfg.Version, err = sarama.ParseKafkaVersion(version); err != nil {
+			return nil, err
+		}
 	}
 	cfg.Producer.Compression = compressionMode
 	cfg.Net.DialTimeout, err = time.ParseDuration(timeout)