标签：err Producer sarama producer kafka CRC config ckafka Compression

初始化的主要代码如下：

	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll // Wait for all in-sync replicas to ack the message
	config.Producer.Retry.Max = 10                   // Retry up to 10 times to produce the message
	config.Producer.Return.Successes = true
	config.Producer.Compression = sarama.CompressionSnappy // Compress messages
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Errors = true
	producer, err := sarama.NewSyncProducer(brokerList, config)
	if err != nil {
		log.Printf("sarama.NewSyncProducer fail:%+v\n", err)
		return err
	}

生产消息的主要代码如下：

	msg := sarama.ProducerMessage{
		Topic: monitorInstance.topic,
		Value: sarama.StringEncoder(resportString),
		//Key: sarama.StringEncoder("monitor_data"),
	}
	partition, offset, err := (*producer).SendMessage(&msg)
	if err != nil {
		log.Printf("kafka SendMessage fail:%+v\n", err)
		return
	}
	log.Printf("Your data is stored with unique identifier important/%d/%d\n", partition, offset)

尝试了以下办法都不行：
1.把producer从async修改为sync
2.增加msg中的key
3.尝试更换topic

最后发现是这个问题：
config.Producer.Compression = sarama.CompressionSnappy
注释掉这一行后正常。

难道就不能加压缩？问了腾讯云的工程师，得到了解决：
config.Version = sarama.V2_1_0_0
config.Producer.Compression = sarama.CompressionSnappy

标签：err,Producer,sarama,producer,kafka,CRC,config,ckafka,Compression
来源： https://www.cnblogs.com/ahfuzhang/p/12737304.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。