反序列化失败的 Kafka 消费者行为答案

【问题标题】：Kafka consumer behaviour on failed deserialization反序列化失败的 Kafka 消费者行为
【发布时间】：2019-04-29 08:38:06
【问题描述】：

我想知道当对象的反序列化失败时如何告诉 kafka 消费者从 kafka 删除记录，记录错误并继续侦听其他传入消息。

我的消费者配置如下：消费者属性：

Map<String, Object> consumerProperties = new HashMap<>();
    consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
            kafkaPropertiesConfiguration.getBootstrapServers());
    consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class.getName());

    consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
            NullDeserializer.class.getName());

    consumerProperties.put(ConsumerConfig.CLIENT_ID_CONFIG,
            kafkaTelecontrolEventConsumerProperties.getClientIdConfig());
    consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG,
            kafkaTelecontrolEventConsumerProperties.getGroupIdConfig());

    return new DefaultKafkaConsumerFactory<>(consumerProperties);

NullDeserializer.class 是我用于测试的反序列化器：

@Log4j2
public class NullDeserializer implements Deserializer<Object> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {

    }

    @Override
    public Object deserialize(String topic, byte[] data) {
        try {
            TeleControl.Event.parseFrom(data) ;
        } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException(e);
        }
        return null;
    }

    @Override
    public void close() {

    }
}

集成流：

return IntegrationFlows
                .from(Kafka
                        .messageDrivenChannelAdapter(telecontrolEventConsumerFactory,
                                eventConsumerProperties.getTelecontrolEventTopic())
                )
                .handle(System.err::println)
                .get();

我的主要问题是解析失败时抛出新的RuntimeException，如何告诉kafkaConsumer记录错误，删除记录并继续处理下一条kafka消息。

【问题讨论】：

标签： java spring apache-kafka spring-integration

【解决方案1】：

参见 2.2 中引入的ErrorHandlingDeserializer。

即将发布的 2.2.1 版本（明天到期）将其替换为类型安全的 ErrorHandlingDeserializer2。

在这两种情况下，反序列化异常都会传递给侦听器容器，而后者又将其直接传递给ErrorHandler，而不是调用侦听器。

【讨论】：