Невозможно опубликовать sh в топи недоставленных писем c при сбое транзакции - PullRequest
Новая
       102

Невозможно опубликовать sh в топи недоставленных писем c при сбое транзакции

0 голосов
/ 20 июня 2020

В своем приложении я настроил Kafka следующим образом. 

@Configuration
public class KafkaConfiguration {
    @Bean
    public ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
                                                                                                 ConsumerFactory<Object, Object> consumerFactory,
                                                                                                 KafkaTransactionManager<?, ?> kafkaTransactionManager,
                                                                                                 KafkaTemplate<Object, Object> template) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, consumerFactory);
        
        // Send all failed message from topic XYZ to another dead-letter topic with .DLT suffix (i.e. XYZ.DLT)
        factory.getContainerProperties().setTransactionManager(kafkaTransactionManager);
        factory.setAfterRollbackProcessor(new DefaultAfterRollbackProcessor<>(new DeadLetterPublishingRecoverer(template), new FixedBackOff(50, 1)));

        return factory;
    }

    @Bean
    public RecordMessageConverter converter() {
      return new StringJsonMessageConverter();
    }
}

Ниже показано, что у меня есть в application.yml

spring:
  kafka:
    admin:
      fail-fast: true
    consumer:
      bootstrap-servers: 127.0.0.1:9092
      group-id: wfo
      enable-auto-commit: false
      auto-offset-reset: latest
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
      properties:
        spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
        spring.deserializer.value.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
        isolation.level: read_committed
    producer:
      bootstrap-servers: 127.0.0.1:9092
      transaction-id-prefix: tx.
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

Чтобы проверить слушателей на сбой транзакции, я создал следующий класс. 

@Service
public class EventListener {
    @KafkaListener(topics="test_topic")
    public void listen(TestEvent event) {
        System.out.println("RECEIVED EVENT: " + event.getPayload());

        if (event.getPayload().contains("fail"))
            throw new RuntimeException("TEST TRANSACTION FAILED");
    }
}

Когда я публикую sh a TestEvent, я вижу полезную нагрузку, напечатанную на консоли. Когда я включаю слово fail в свою полезную нагрузку, выдается RuntimeException, и я вижу сообщение об ошибке Transaction rolled back на консоли.

Однако после неудачной попытки примерно через минуту я увидел сообщение об ошибке следующее исключение на консоли. 

2020-06-20 17:07:46,326 ERROR [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] o.s.kafka.support.LoggingProducerListener   : Exception thrown when sending a message with key='null' and payload='{"customerCode":"DVTPRDFT411","payload":"MSGfail 90"}' to topic test_topic.DLT and partition 2:
org.apache.kafka.common.errors.TimeoutException: Topic test_topic.DLT not present in metadata after 60000 ms.
2020-06-20 17:07:46,327 ERROR [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] o.s.kafka.listener.DeadLetterPublishingRecoverer   : Dead-letter publication failed for: ProducerRecord(topic=test_topic.DLT, partition=2, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 102, 116, 46, 100, 101, 109, 111, 46, 100, 116, 111, 46, 100, 101, 98, 117, 103, 46, 75, 97, 102, 107, 97, 69, 118, 101, 110, 116, 82, 101, 113, 117, 101, 115, 116, 36, 71, 101, 110, 101, 114, 105, 99, 69, 118, 101, 110, 116]), RecordHeader(key = kafka_dlt-original-topic, value = [116, 101, 115, 116, 95, 116, 111, 112, 105, 99]), RecordHeader(key = kafka_dlt-original-partition, value = [0, 0, 0, 2]), RecordHeader(key = kafka_dlt-original-offset, value = [0, 0, 0, 0, 0, 0, 0, 0]), RecordHeader(key = kafka_dlt-original-timestamp, value = [0, 0, 1, 114, -48, -29, -90, 47]), RecordHeader(key = kafka_dlt-original-timestamp-type, value = [67, 114, 101, 97, 116, 101, 84, 105, 109, 101]), RecordHeader(key = kafka_dlt-exception-fqcn, value = [111, 114, 103, 46, 115, 112, 114, 105, 110, 103, 102, 114, 97, 109, 101, 119, 111, 114, 107, 46, 107, 97, 102, 107, 97, 46, 108, 105, 115, 116, 101, 110, 101, 114, 46, 76, 105, 115, 116, 101, 110, 101, 114, 69, 120, 101, 99, 117, 116, 105, 111, 110, 70, 97, 105, 108, 101, 100, 69, 120, 99, 101, 112, 116, 105, 111, 110]), RecordHeader(key = kafka_dlt-exception-message, value = [76, 105, 115, 116, 101, 110, 101, 114, 32, 109, 101, 116, 104, 111, 100, 32, 39, 112, 117, 98, 108, 105, 99, 32, 118, 111, 105, 100, 32, 99, 111, 109, 46, 102, 116, 46, 100, 101, 109, 111, 46, 115, 101, 114, 118, 105, 99, 101, 46, 69, 118, 101, 110, 116, 76, 105, 115, 116, 101, 110, 101, 114, 46, 108, 105, 115, 116, 101, 110, 40, 99, 111, 109, 46, 102, 116, 46, 101, 118, 101, 110, 116, 46, 109, 97, 110, 97, 103, 101, 109, 101, 110, 116, 46, 100, 101, 102, 105, 110, 105, 116, 105, 111, 110, 46, 84, 101, 115, 116, 69, 118, 101, 110, 116, 41, 39, 32, 116, 104, 114, 101, 119, 32, 101, 120, 99, 101, 112, 116, 105, 111, 110, 59, 32, 110, 101, 115, 116, 101, 100, 32, 101, 120, 99, 101, 112, 116, 105, 111, 110, 32, 105, 115, 32, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 82, 117, 110, 116, 105, 109, 101, 69, 120, 99, 101, 112, 116, 105, 111, 110, 58, 32, 84, 69, 83, 84, 32, 84, 82, 65, 78, 83, 65, 67, 84, 73, 79, 78, 32, 70, 65, 73, 76, 69, 68, 59, 32, 110, 101, 115, 116, 101, 100, 32, 101, 120, 99, 101, 112, 116, 105, 111, 110, 32, 105, 115, 32, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 82, 117, 110, 116, 105, 109, 101, 69, 120, 99, 101, 112, 116, 105, 111, 110, 58, 32, 84, 69, 83, 84, 32, 84, 82, 65, 78, 83, 65, 67, 84, 73, 79, 78, 32, 70, 65, 73, 76, 69, 68]), RecordHeader(key = kafka_dlt-exception-stacktrace, value = [111, 100]), RecordHeader(key = b3, value = [100, 57, 51, 100, 56, 97, 98, 57, 55, 54, 97, 100, 54, 102, 49, 100, 45, 100, 57, 51, 100, 56, 97, 98, 57, 55, 54, 97, 100, 54, 102, 49, 100, 45, 49])], isReadOnly = false), key=null, value={"customerCode":"DVTPRDFT411","payload":"MSGfail 90"}, timestamp=null)
org.springframework.kafka.KafkaException: Send failed; nested exception is org.apache.kafka.common.errors.TimeoutException: Topic test_topic.DLT not present in metadata after 60000 ms.
    at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:570)
    at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:385)
    at org.springframework.kafka.listener.DeadLetterPublishingRecoverer.publish(DeadLetterPublishingRecoverer.java:278)
    at org.springframework.kafka.listener.DeadLetterPublishingRecoverer.lambda$accept$3(DeadLetterPublishingRecoverer.java:209)
    at org.springframework.kafka.core.KafkaTemplate.executeInTransaction(KafkaTemplate.java:463)
    at org.springframework.kafka.listener.DeadLetterPublishingRecoverer.accept(DeadLetterPublishingRecoverer.java:208)
    at org.springframework.kafka.listener.DeadLetterPublishingRecoverer.accept(DeadLetterPublishingRecoverer.java:54)
    at org.springframework.kafka.listener.FailedRecordTracker.skip(FailedRecordTracker.java:106)
    at org.springframework.kafka.listener.SeekUtils.lambda$doSeeks$2(SeekUtils.java:84)
    at java.util.ArrayList.forEach(Unknown Source)
    at org.springframework.kafka.listener.SeekUtils.doSeeks(SeekUtils.java:81)
    at org.springframework.kafka.listener.DefaultAfterRollbackProcessor.process(DefaultAfterRollbackProcessor.java:102)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.recordAfterRollback(KafkaMessageListenerContainer.java:1700)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListenerInTx(KafkaMessageListenerContainer.java:1662)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1614)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:1348)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1064)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:972)
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
    at java.util.concurrent.FutureTask.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)

По некоторым причинам код не может публиковать sh события в DLT topi c. Я попытался отредактировать конфигурации брокера, включив в него listeners=PLAINTEXT://:9092, но это не помогло.

Буду очень признателен, если вы укажете мне направление решения этой проблемы.

* * ОБНОВЛЕНИЕ:

Я вижу DLT topi c, когда выполняю команду kafka-topics --list --zookeeper localhost:2181.

enter image description here

By the way, this might be a bug from the latest Spring Kafka version. I downgraded to Kafka v2.4.1 and Spring Boot v2.2.7 and it's working fine. I reported the bug here.

https://github.com/spring-projects/spring-kafka/issues/1516

Если в Kafka v2.5.0 нужно выполнить какие-то новые настройки, чтобы эта работа работала, дайте мне знать.

1 Ответ

2 голосов
/ 20 июня 2020

org. apache .kafka.common.errors.TimeoutException: Topi c test_topi c .DLT отсутствует в метаданных после 60000 мс.

Из ошибки , похоже, что у вас нет test_topic.DLT.
Для каждого созданного вами topi c, например XXX, вам нужно создать его соответствующий XXX.DLT topi c.

В качестве примечания, ErrorHandlingDeserializer2 является устаревшим с 2.5.

ОБНОВЛЕНИЕ:
Когда вы перечисляете темы kafka-topics.bat --bootstrap-server 127.0.0.1:9092 --list, можете ли вы увидеть test_topic.DLT? И когда вы описываете его kafka-topics.bat --bootstrap-server 127.0.0.1:9092 --describe --topic test_topic.DLT, можете ли вы увидеть то же количество разделов, что и исходный topi c, точнее вы видите раздел 2?

Согласно Spring Kafka Documentation :

По умолчанию запись недоставленного письма отправляется в топи c с именем .DLT (исходное имя топи c с суффиксом .DLT) и в тот же раздел, что и исходная запись.

...