SeekToCurrentErrorHandler: DeadLetterPublishingRecoverer не обрабатывает ошибки десериализации - PullRequest
Новая
       1

SeekToCurrentErrorHandler: DeadLetterPublishingRecoverer не обрабатывает ошибки десериализации

1 голос
/ 24 июня 2019

Я пытаюсь написать потребитель kafka, используя библиотеку spring-kafka версии 2.3.0.M2.Для обработки ошибок во время выполнения я использую SeekToCurrentErrorHandler.class с DeadLetterPublishingRecoverer в качестве моего восстановителя.Это работает нормально только тогда, когда мой потребительский код выдает исключение, но не удается, если не удается десериализовать сообщение.

Я попытался реализовать ErrorHandler сам, и мне это удалось, но с этим подходом я сам заканчиваю тем, что писал код DLT для обработки сообщений об ошибкахчто я не хочу делать.

Ниже приведены мои свойства кафки 

spring:
   kafka:
     consumer:
        bootstrap-servers: localhost:9092
        group-id: group_id
        auto-offset-reset: latest
        key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
        value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
        properties:
          spring.json.trusted.packages: com.mypackage
          spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
          spring.deserializer.value.delegate.class: org.apache.kafka.common.serialization.StringDeserializer

public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ConsumerFactory<Object, Object> kafkaConsumerFactory,
        KafkaTemplate<Object, Object> template) {
      ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
      configurer.configure(factory, kafkaConsumerFactory);
    factory.setErrorHandler(new SeekToCurrentErrorHandler(new DeadLetterPublishingRecoverer(template), maxFailures));}

1 Ответ

0 голосов
/ 24 июня 2019

У меня нормально работает (учтите, что Boot автоматически настроит обработчик ошибок) ... 

@SpringBootApplication
public class So56728833Application {

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

    @Bean
    public SeekToCurrentErrorHandler errorHandler(KafkaTemplate<String, String> template) {
        SeekToCurrentErrorHandler eh = new SeekToCurrentErrorHandler(new DeadLetterPublishingRecoverer(template), 3);
        eh.setClassifier( // retry for all except deserialization exceptions
                new BinaryExceptionClassifier(Collections.singletonList(DeserializationException.class), false));
        return eh;
    }

    @KafkaListener(id = "so56728833"
            + "", topics = "so56728833")
    public void listen(Foo in) {
        System.out.println(in);
        if (in.getBar().equals("baz")) {
            throw new IllegalStateException("Test retries");
        }
    }

    @KafkaListener(id = "so56728833dlt", topics = "so56728833.DLT")
    public void listenDLT(Object in) {
        System.out.println("Received from DLT: " + (in instanceof byte[] ? new String((byte[]) in) : in));
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so56728833").partitions(1).replicas(1).build();
    }

    @Bean
    public NewTopic dlt() {
        return TopicBuilder.name("so56728833.DLT").partitions(1).replicas(1).build();
    }

    public static class Foo {

        private String bar;

        public Foo() {
            super();
        }

        public Foo(String bar) {
            this.bar = bar;
        }

        public String getBar() {
            return this.bar;
        }

        public void setBar(String bar) {
            this.bar = bar;
        }

        @Override
        public String toString() {
            return "Foo [bar=" + this.bar + "]";
        }

    }

}

spring:
  kafka:
    consumer:
      auto-offset-reset: earliest
      enable-auto-commit: false
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
      properties:
        spring.json.trusted.packages: com.example
        spring.deserializer.key.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        spring.json.value.default.type: com.example.So56728833Application$Foo
    producer:
      key-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

logging:
  level:
    org.springframework.kafka: trace

У меня 3 записи в теме: 

"badJSON"
"{\"bar\":\"baz\"}"
"{\"bar\":\"qux\"}"

Я вижу, что первый идет прямо в DLT, а второй идет туда после 3 попыток.

...