推荐一款非常好用的kafka管理平台,kafka的灵魂伴侣 滴滴开源Logi-KafkaManager 一站式Kafka监控与管控平台
技术交流
有想进滴滴LogI开源用户群的加我个人微信: jjdlmn_ 进群(备注：进群) 群里面主要交流 kakfa、es、agent、LogI-kafka-manager、等等相关技术; 群内有专人解答你的问题 对~ 相关技术领域的解答人员都有; 你问的问题都会得到回应
有想进 滴滴LogI开源用户群 的加我个人微信: jjdlmn_ 进群(备注：进群) 群里面主要交流 kakfa、es、agent、以及其他技术 群内有专人解答疑问,你所问的都能得到回应
CORRUPT_MESSAGE
这个错误一般是压缩策略为cleanup.policy=compact的情况下，key不能为空
代码语言：javascript复制o.a.k.c.p.i.Sender 595 [WARN] [Producer clientId=producer-1] Got error produce response with correlation id
131 on topic-partition SHI_TOPIC1-0, retrying (2147483521 attempts left). Error: CORRUPT_MESSAGE查看一下压缩策略
代码语言：javascript复制bin/kafka-topics.sh --describe --zookeeper xxxx:2181 --topic SHI_TOPIC1
Topic:SHI_TOPIC1 PartitionCount:1 ReplicationFactor:1 Configs:cleanup.policy=compact
Topic: SHI_TOPIC1 Partition: 0 Leader: 0 Replicas: 0 Isr: 0Configs:cleanup.policy=compact :
然后再检查一下自己发送消息的时候是不是没有传 key
参考链接
问题堆栈信息
代码语言：javascript复制org.springframework.kafka.listener.ListenerExecutionFailedException: invokeHandler Failed;
nested exception is java.lang.IllegalStateException: No Acknowledgment available as an argument,
the listener container must have a MANUAL AckMode to populate the Acknowledgment.;
nested exception is java.lang.IllegalStateException:
No Acknowledgment available as an argument, the listener container must have a MANUAL AckMode to populate the Acknowledgment.问题原因
解决方案
问题堆栈信息
代码语言：javascript复制 Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; nested exception is java.lang.IllegalStateException: Consumer cannot be configured for auto commit for ackMode MANUAL_IMMEDIATE问题原因
不能再配置中既配置
kafka.consumer.enable-auto-commit=true自动提交; 然后又在监听器中使用手动提交
例如:
代码语言：javascript复制kafka.consumer.enable-auto-commit=true @Autowired
private ConsumerFactory consumerFactory;
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaManualAckListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
//设置提交偏移量的方式 当Acknowledgment.acknowledge()侦听器调用该方法时，立即提交偏移量
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
return factory;
}
/**
* 手动ack 提交记录
* @param data
* @param ack
* @throws InterruptedException
*/
@KafkaListener(id = "consumer-id2",topics = "SHI_TOPIC1",concurrency = "1",
clientIdPrefix = "myClientId2",containerFactory = "kafkaManualAckListenerContainerFactory")
public void consumer2(String data, Acknowledgment ack) {
log.info("consumer-id2-手动ack,提交记录,data:{}",data);
ack.acknowledge();
}解决方法:
代码语言：javascript复制将自动提交关掉,或者去掉手动提交; 如果你想他们都同时存在,某些情况自动提交；某些情况手动提交; 那你创建 一个新的
consumerFactory将它的是否自动提交设置为false；比如
@Configuration
@EnableKafka
public class KafkaConfig {
@Autowired
private KafkaProperties properties;
/**
* 创建一个新的消费者工厂
* 创建多个工厂的时候 SpringBoot就不会自动帮忙创建工厂了；所以默认的还是自己创建一下
* @return
*/
@Bean
public ConsumerFactory<Object, Object> kafkaConsumerFactory() {
Map<String, Object> map = properties.buildConsumerProperties();
DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>( map);
return factory;
}
/**
* 创建一个新的消费者工厂
* 但是修改为不自动提交
*
* @return
*/
@Bean
public ConsumerFactory<Object, Object> kafkaManualConsumerFactory() {
Map<String, Object> map = properties.buildConsumerProperties();
map.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>( map);
return factory;
}
/**
* 手动提交的监听器工厂 (使用的消费组工厂必须 kafka.consumer.enable-auto-commit = false)
* @return
*/
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaManualAckListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(kafkaManualConsumerFactory());
//设置提交偏移量的方式 当Acknowledgment.acknowledge()侦听器调用该方法时，立即提交偏移量
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
return factory;
}
/**
* 监听器工厂 批量消费
* @return
*/
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> batchFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(kafkaConsumerFactory());
factory.setBatchListener(true);
return factory;
}
}消费者监听的时候 指定对应的 容器工厂就行了kafkaManualAckListenerContainerFactory
/**
* 手动ack 提交记录
* @param data
* @param ack
* @throws InterruptedException
*/
@KafkaListener(id = "consumer-id2",topics = "SHI_TOPIC1",concurrency = "1",
clientIdPrefix = "myClientId2",containerFactory = "kafkaManualAckListenerContainerFactory")
public void consumer2(String data, Acknowledgment ack) {
log.info("consumer-id2-手动ack,提交记录,data:{}",data);
ack.acknowledge();
}问题堆栈信息
代码语言：javascript复制[WARN] Error registering AppInfo mbean
javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=myClientId-3问题原因
官网描述 The client.id property (if set) is appended with -n where n is the consumer instance that corresponds to the concurrency. This is required to provide unique names for MBeans when JMX is enabled. 意思是这个id在JMX中注册需要id名唯一；不要重复了;
解决方法:
将监听器的id修改掉为唯一值 或者 消费者的全局配置属性中不要知道 client-id ；则系统会自动创建不重复的client-id
欢迎 Star和 共建由 滴滴开源的kafka的管理平台,非常优秀非常好用的一款kafka管理平台
满足所有开发运维日常需求
滴滴开源Logi-KafkaManager 一站式Kafka监控与管控平台
