使用Spring Cloud Stream处理Avro反序列化异常

Question

我有一个使用Spring Cloud Stream和Spring Kafka的应用程序，它处理Avro消息。该应用程序运行良好，但现在我想添加一些错误处理。

目标是：捕获反序列化异常，使用异常详细信息原始Kafka消息自定义上下文信息构建新对象，并将此对象推送到专用Kafka主题。基本上是DLQ，但原始消息将被截获并修饰。

问题是：虽然我可以拦截异常，但我不知道如何从Kafka那里获取原始消息（下面的TODO 1）。我已经查看了ConsumerAwareErrorHandler中返回的数据对象。把手，我在那里看不到。

以下是我的代码：

@EnableBinding(EventStream.class)
@SpringBootApplication
@Slf4j
public class SpringcloudApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringcloudApplication.class, args);
    }

    /* Configure custom exception handler */
    @Bean
    public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> cust() {
        return (container, destination, group) -> {
            container.setErrorHandler(new ConsumerAwareErrorHandler() {
                @Override
                public void handle(Exception thrownException, ConsumerRecord<?, ?> data, Consumer<?, ?> consumer) {
                    log.info("Got error with data: {}", data);
                    // TODO 1 - How to get original message?
                    // TODO 2 - Send to dedicated (DLQ) topic
                }
            });
        };
    }

    @StreamListener(EventStream.INBOUND)
    public void consumeEvent(@Payload Message message) {
        log.info("Consuming event --> {}", message.toString());
        produceEvent(message);
    }

    @Autowired private EventStream eventStream;
    public Boolean produceEvent(Message message) {
        log.info("Producing event --> {}", message.toString());
        return eventStream
                .producer()
                .send(MessageBuilder.withPayload(message)
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                    .build());
    }

}

以及属性文件：

spring:
  cloud:
    stream:
      default-binder: kafka
      default:
        consumer:
          useNativeEncoding: true
        producer:
          useNativeEncoding: true
      kafka:
        binder:
          brokers: localhost:9092
          producer-properties:
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
            schema.registry.url: "http://localhost:8081"
          consumer-properties:
            key.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
            value.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
            schema.registry.url: "http://localhost:8081"
            specific.avro.reader: true
            spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
            spring.deserializer.value.delegate.class: io.confluent.kafka.serializers.KafkaAvroDeserializer
      bindings:
        event-consumer:
          destination: data_stream_in # incoming topic
          contentType: application/**avro
          group: data_stream_consumer
        event-producer:
          destination: data_stream_out
          contentType: application/**avro

我正在使用以下版本：

• Spring防尘套2.3.2。释放

感谢您的帮助！

Answer 1

句柄方法中的第二个参数是消费者记录，它是原始的Kafka记录，但如果您希望记录自动发送到DLQ，您可以执行以下操作。

@Bean
    public ListenerContainerCustomizer<AbstractMessageListenerContainer<byte[], byte[]>> customizer(SeekToCurrentErrorHandler errorHandler) {
        return (container, dest, group) -> {
            container.setErrorHandler(errorHandler);
        };
    }



    @Bean
    public SeekToCurrentErrorHandler errorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {
        return new SeekToCurrentErrorHandler(deadLetterPublishingRecoverer);
    }


    @Bean
    public DeadLetterPublishingRecoverer publisher(KafkaOperations bytesTemplate) {
        return new DeadLetterPublishingRecoverer(bytesTemplate);
    }

实际上，您正在设置一个SeekToCurrentErrorHandler，它能够将失败的记录发送到DLQ。有关死信发布恢复器（DeadLetterPublishingRecoverer）工作原理的更多详细信息，请参阅Apache Kafka的Spring参考文档：https://docs.spring.io/spring-kafka/docs/current/reference/html/#dead-信件；有关详细信息，请参阅文档交互处理程序：https://docs.spring.io/spring-kafka/docs/current/reference/html/#seek-至当前

您还需要配置和ErrorHandlingDeserializer，

spring.cloud.stream.kafka.binder.configuration.value.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
spring.cloud.stream.kafka.binder.configuration.spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
...
similar for the value class. 

有关ErrorHandlingDeserializer的更多信息：https://docs.spring.io/spring-kafka/docs/current/reference/html/#error-handling-deserializer

如果要修改记录并将自定义消息添加到DLQ，可以通过覆盖handle方法，然后访问ConsumerRecord，然后调用super class方法来实现。