【kafka异常】使用Spring-kafka遇到的坑

痛定思痛。 2022-12-22 04:35 199阅读 0赞

> 推荐一款非常好用的kafka管理平台,kafka的灵魂伴侣  
>   
> [滴滴开源Logi-KafkaManager 一站式Kafka监控与管控平台][Logi-KafkaManager _Kafka]

--------------------

## 技术交流 ##

有想进**滴滴LogI开源用户群**的加我个人微信: jjdlmn\_ 进群(备注：进群)  
群里面主要交流 **kakfa**、**es**、**agent**、**LogI-kafka-manager**、等等相关技术;  
群内有专人解答你的问题  
对~ 相关技术领域的解答人员都有; 你问的问题都会得到回应

有想进 滴滴LogI开源用户群 的加我个人微信: jjdlmn\_ 进群(备注：进群)  
群里面主要交流 kakfa、es、agent、以及其他技术  
群内有专人解答疑问,你所问的都能得到回应

--------------------

### CORRUPT\_MESSAGE ###

**这个错误一般是压缩策略为cleanup.policy=compact的情况下，key不能为空**

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

查看一下压缩策略

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: 0

`Configs:cleanup.policy=compact` :

然后再检查一下自己发送消息的时候是不是没有传 `key`

[参考链接][Link 1]

**问题堆栈信息**

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.

**问题原因**

**解决方案**

**问题堆栈信息**

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` 自动提交; 然后又在监听器中使用手动提交**

例如:

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();
    
        }

**解决方法:**

> 将自动提交关掉,或者去掉手动提交;  
> 如果你想他们都同时存在,某些情况自动提交；某些情况手动提交; 那你创建 一个新的  
> `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();
    
        }

--------------------

**问题堆栈信息**

[WARN] Error registering AppInfo mbean
     javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=myClientId-3

**问题原因**

> [官网描述 ][Link 2]  
> 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监控与管控平台][Logi-KafkaManager _Kafka]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center]

[Logi-KafkaManager _Kafka]: https://github.com/didi/LogiKM
[Link 1]: https://www.orchome.com/6750
[Link 2]: https://docs.spring.io/spring-kafka/docs/2.5.8.RELEASE/reference/html/#message-listener-container
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/3040fd05e9c54050af0d58efee731be9.png