Spring Kafka批量反序列化错误处理及手动提交偏移量问题

问题1：批量确认/提交偏移量的流程解析 #

在MANUAL_IMMEDIATE确认模式下，Kafka容器的批量确认逻辑是这样的：

• 当你在监听器中调用Acknowledgment.acknowledge()时，容器会异步提交当前批次所有消息的偏移量到Kafka。
• 但如果批次处理抛出异常，控制权会转到你配置的BatchErrorHandler。这时候要注意两个关键点：
  1. isAckAfterHandle()返回true时，容器会在ErrorHandler执行完毕后自动提交整个批次的偏移量——这显然不是你想要的，因为你只想跳过失败的那条消息，继续处理后面的内容。
  2. 直接调用consumer.commitSync()不生效的核心原因：容器在MANUAL_IMMEDIATE模式下会维护自己的偏移量跟踪机制，手动提交会被容器的逻辑覆盖，甚至可能引发偏移量不一致的问题。

正确的思路是：跳过失败的消息，手动将偏移量移动到失败消息的下一个位置，让容器继续处理后续批次，而不是强行提交整个批次的偏移量。

问题2：获取反序列化失败的原始消息 #

你遇到data为null的情况，是因为当ErrorHandlingDeserializer在反序列化阶段失败时，整个批次的ConsumerRecords可能无法正常传递到BatchErrorHandler，或者失败记录的value会被置为null。

想要可靠获取原始消息，有两种实用方式：

1. 从SerializationException中解析关键信息：你已经在尝试解析异常消息，但可以优化解析逻辑，更稳健地提取分区、偏移量等信息，再通过consumer.seek()跳过失败消息。
2. 配置ErrorHandlingDeserializer保留原始字节：在消费者配置中添加特定参数，让反序列化失败时保留原始的字节数据，这样可以从异常的cause中拿到DeserializationException，进而获取未序列化的原始byte[]。

完整修改后的代码示例 #

1. 更新消费者配置，保留原始反序列化失败的字节 #

@Bean 
public Map<String, Object> consumerConfigs() { 
    Map<String, Object> props = new HashMap<>(); 
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); 
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords); 
    props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollIntervalMsConfig); 
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); 
    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); 
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 
    // 配置ErrorHandlingDeserializer基础参数
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class); 
    props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class); 
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class); 
    props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class); 
    // 添加这行，让反序列化失败时保留原始字节数据
    props.put(ErrorHandlingDeserializer.VALUE_FUNCTION, (ctx, e) -> ctx.value());
    return props; 
} 

2. 优化BatchErrorHandler，正确跳过失败消息并送入死信队列 #

@Bean(KAFKA_LISTENER) 
public ConcurrentKafkaListenerContainerFactory<String, MyDTO> kafkaListenerContainerFactory() { 
    ConcurrentKafkaListenerContainerFactory<String, MyDTO> factory = new ConcurrentKafkaListenerContainerFactory<>(); 
    factory.setConsumerFactory(consumerFactory()); 
    factory.setBatchListener(true); 
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); 
    factory.setBatchErrorHandler(new ConsumerAwareBatchErrorHandler() { 
        @Override 
        public void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer) { 
            if (thrownException instanceof SerializationException) { 
                SerializationException se = (SerializationException) thrownException; 
                String errorMsg = se.getMessage(); 
                try { 
                    // 解析异常消息，提取主题、分区、偏移量
                    String errorSegment = errorMsg.split("Error deserializing key/value for partition ")[1]; 
                    String topic = errorSegment.split("-")[0]; 
                    String partitionOffsetSegment = errorSegment.split("-")[1]; 
                    int partition = Integer.parseInt(partitionOffsetSegment.split(" at")[0]); 
                    int offset = Integer.parseInt(partitionOffsetSegment.split("offset ")[1]); 
                    
                    // 跳过当前失败的消息，将偏移量移动到下一个位置
                    TopicPartition topicPartition = new TopicPartition(topic, partition); 
                    consumer.seek(topicPartition, offset + 1); 
                    
                    // 获取原始消息并送入错误队列（如果配置了VALUE_FUNCTION）
                    if (se.getCause() instanceof DeserializationException) { 
                        DeserializationException de = (DeserializationException) se.getCause(); 
                        byte[] rawValue = de.getData(); 
                        String rawKey = de.getKey() != null ? new String(de.getKey()) : null; 
                        sendToErrorQueue(topic, partition, offset, rawKey, rawValue); 
                    } 
                } catch (Exception e) { 
                    log.error("Failed to parse serialization error details", e); 
                } 
            } 
            // 注意：不要调用consumer.commitSync()，MANUAL_IMMEDIATE模式下容器会处理后续的偏移量提交
        } 

        @Override 
        public boolean isAckAfterHandle() { 
            // 返回false，避免容器提交整个批次的偏移量（我们已经手动跳过了失败消息）
            return false; 
        } 
    }); 
    return factory; 
}

// 示例：将失败消息发送到错误队列的方法
private void sendToErrorQueue(String originalTopic, int partition, long offset, String key, byte[] value) {
    // 这里替换成你的KafkaTemplate发送逻辑
    kafkaTemplate.send("error-topic", key, value);
    log.warn("Sent failed message to error queue: topic={}, partition={}, offset={}", originalTopic, partition, offset);
}

关键注意事项 #

• isAckAfterHandle()返回false是核心：避免容器提交整个批次的偏移量，因为我们已经通过seek()手动跳过了失败的消息。
• 解析异常消息时要做好异常捕获：防止因为Kafka异常消息格式变化导致解析失败，引发新的问题。
• 保留原始字节的配置可以让你完整获取失败消息的原始数据，方便后续的问题排查与处理。

内容的提问来源于stack exchange，提问作者Vikas Bansal