Spring Kafka消费者如何从Avro反序列化程序异常中跳过

Question

我正在使用Spring Kafka consumer和Avro模式构建我的应用程序。

但是，如果消息无法反序列化到我构建的指定Avro特定记录，消费者将不断地反复尝试相同的消息（无限重试）。

在这种情况下，如果我的使用者出现反序列化程序异常，我如何配置使用者应用程序以跳过当前消息并移动到下一个偏移量。

我已经研究了Spring Kafka错误句柄，它只能处理侦听器中的异常，而不是在反序列化阶段。

我的消费者应用程序非常简单：

@KafkaListener(id = "demo-consumer-stream-group", topics = "customer-output-")
  public void process(ConsumerRecord<String, Customer> record) {
    LOGGER.info("Customer key: {} and value: {}", record.key(), record.value());
    LOGGER.info("topic: {}, partition: {}, offset: {}", record.topic(), record.partition(), record.offset());
  }

基于此代码，有时接收到的消息可能无法反序列化到正确的Customer对象。

此外，我看到最近的一个解决方案是使用Spring Kafka的ErrorHandlingDeserializer2来处理这个问题，但既然我使用的是Kafkavrodeserializer我该如何计算这些配置呢？我目前的配置是：

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);

Answer 1

留档里有解释。

通过custom Spring属性将反序列化程序设置为错误处理反序列化程序及其委托。

您可以使用DefaultKafkaConsumerFactory构造函数，该构造函数接受键和值反序列化器对象，并连接到已配置了适当委托的相应ErrorHandlingDeserializer2实例中。或者，您可以使用使用者配置属性（ErrorHandlingDeserializer使用这些属性）来实例化委托。属性名为ErrorHandlingDeserializer2。键_反序列化器_类和错误处理反序列化器2。值\反序列化程序\类。属性值可以是类或类名。以下示例显示了如何设置这些属性：

... // other props
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, JsonDeserializer.class);
props.put(JsonDeserializer.KEY_DEFAULT_TYPE, "com.example.MyKey")
props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "com.example.MyValue")
props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example")
return new DefaultKafkaConsumerFactory<>(props);

Answer 2

您需要将当前值键反序列化器设置为ErrorHandlingDeserializer.class并将当前值键反序列化器设置为ErrorHandlingDeserializer Key/Value反序列化器属性

这将看起来类似于这样的东西：

... // other props
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, KafkaAvroDeserializer.class);
props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, StringDeserializer.class);
return new DefaultKafkaConsumerFactory<>(props);