{%hackmd BJzAwtWvp %}

Kafka 例外處理

首先，原生kafka是不支持消息重試的。但是spring kafka 2.7+ 封裝了 Retry Topic這個功能。 請各位再嘗試的時候，請注意個依賴的版本問題，小弟我啟環境的時候，試了好多遍Orz。

專案位置

https://github.com/mister33221/spring-kafka-error-handling-example.git

環境

  • java: openjdk-17
  • spring-boot: 2.6.6
  • spring-kafka: 2.8.5
  • lombok: 1.2.11

開始弄髒手

其實我都寫好了，應該不會弄很髒~

  1. 把這個寫好的範例專案clone下來

    https://github.com/mister33221/spring-kafka-error-handling-example.git

  2. 先使用內含的docker-compose.yml搭配docker將kafka、zookeeper、broker、control-center起起來，在cmd介面中輸入以下指令

    docker-compose up

    kafka的控制平台port號為9021。 成功完成後你可以在網址輸入localhost:9021，如果出現以下畫面就是成功了。

  3. 接著就更新pom.xml把依賴載回來

  4. 啟動專案

  5. 完成啦!! 下課!!

如何測試

程式碼中我盡量都有寫上註解，以更好了解做了甚麼事情。

  • 程式中有分為三種不同的處理方式

    1. blocking: 會鎖在同一個topic上，並在同一個topic上繼續retry
    2. non-blocking: 會把topic放開，並新開topic，在這上面進行retry
      1. multiple Topic: 每次retry，都新開一個topic
      2. single Topic: 第一次retry會

  • 首先看到KafkaConfig.java，這裡定義了error處理的方式

    // @EnableKafka: Enables detection of @KafkaListener annotations on Spring-managed beans
@EnableKafka
// @Configuration: Indicates that a class declares one or more @Bean methods and may be processed by the Spring container to generate bean definitions and service requests for those beans at runtime
@Configuration
public class KafkaConfig {

    // KafkaTemplate: A template for sending messages to Kafka topics
    private final KafkaTemplate<Object, Object> template;

    // ConsumerFactory: A factory for creating Kafka consumers
    private final ConsumerFactory<String, String> consumerFactory;

    public KafkaConfig(KafkaTemplate<Object, Object> template, ConsumerFactory<String, String> consumerFactory) {
        this.template = template;
        this.consumerFactory = consumerFactory;
    }

    // Container Factory containing bocking error handler
    // ConcurrentKafkaListenerContainerFactory: A factory for creating Kafka listener containers with a concurrency
    // of 1 (i.e. a single thread) and a batch size of 1 (i.e. a single record per poll)
    // kafkaBlockingRetryContainerFactory: A bean that creates a container factory for blocking retry
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaBlockingRetryContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // for blocking retry
        factory.setCommonErrorHandler(retryErrorHandler());
        consumerFactory.getListeners();
        return factory;
    }

    // For blocking retry
    // This is a blocking retry (will move offset only when all tries are completed) error handler configured with
    // DeadLetterPublishingRecoverer which publishes event to DLT when tries are over
    // retryErrorHandler: A bean that handles errors in the listener container
    public DefaultErrorHandler retryErrorHandler() {
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
        return new DefaultErrorHandler(recoverer, new FixedBackOff(1000, 3));
    }

    // the kafka version is 2.8.3, so the bean name is RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME
    // destinationTopicResolver: A bean that resolves the destination topic for a given retry topic
    @Bean(name = RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME)
    public DestinationTopicResolver destinationTopicResolver(ApplicationContext context) {
        DefaultDestinationTopicResolver resolver = new DefaultDestinationTopicResolver(Clock.systemUTC(), context);
        resolver.setClassifications(Collections.emptyMap(), true);
        return resolver;
    }



}

  • 來看 Consumer.java

    @Slf4j
@Component
public class Consumer {

    private final KafkaTemplate<String, String> template;

    public Consumer(KafkaTemplate<String, String> template) {
        this.template = template;
    }

    // @KafkaListener: It is used to listen to the messages from the specified topic.
    // listen: This method is used to listen to the messages from the Main Topic.
    // ConsumerRecord: It is used to get the message from the topic.
    // @Header: It is used to get the header information from the message.
    // KafkaHeaders.RECEIVED_TOPIC: It is used to get the topic name from the header.
    @KafkaListener(topics = "products")
    public void listen(ConsumerRecord<String, String> message, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        try {

            if (message.key().equals("product2")) {
                throw new RuntimeException("Exception in main consumer");
            }
            log.info("message consumed - key: {} , value: {}, at: {}", message.key(), message.value());
        } catch (Exception e) {
            log.error("failed to consume - key: {}", message.key());
            // send failed event to another retry topic
            // If you want to test no-bloking retry, you have to comment out code which are marked "For blocking retry" in KafkaConfig.java
            template.send("blocking-products-retry", message.key(), message.value());
//            template.send("non-bloking-multipleTopicRetryConsumer-products-retry", message.key(), message.value());
//            template.send("non-bloking-singleTopicRetryConsumer-products-retry", message.key(), message.value());
        }

    }

}

    • 倒數三行為三種不同處理方式，可透過註解來選用。如果要使用blocking的話，要到config中將我有標注for blocking retry的部分註解起來。

      1. 我們使用Control Centeter-localhost:9021來進行測試
      2. 點擊topic
      3. 找到 product 這個 topic -> messages -> produce a message to this topic
      4. 在key輸入 1 或 null 就可以製造exception

      • blocking 我們可以看下圖，我只截了第一張，你會發現每次就噴一個exception，且每一次都是同一個原始的 topic ，因為他已經被鎖在上面了，造成後面連續的噴錯，而這個重試的預設機制為 retry 3次 ，每次間隔1秒。也可以到config的檔案中自行設定。

      • non-block-multipleTopicRetryConsumer 我們可以看下面的圖，發現他每次都在 topic 名稱後加上 retry 及編號，顯示出他把上一次用的 topic 放開了，並用一個全新的 topic 來進行 retry 。

      • non-blok-singleTopicRetryConsumer-products-retry 我們可以看下面的圖，發現他開始要 retry 時，在原本的 topic 後加上 retry ，並每一次的retry都使用這個新建的 topic 來進行。

使用retry的影響

使用 Spring 提供的 RetryTopic 功能，對 Apache Kafka 中的 topic 數量有以下影響：

  • 增加存儲需求：使用 RetryTopic 功能會將重試的消息存儲在 topic 中，因此存儲需求會增加。
  • 增加消息複雜度：使用 RetryTopic 功能會導致消息的複雜性增加，因此需要更強大的系統和網絡帶寬。
  • 增加監控和日誌分析負擔：使用 RetryTopic 功能會增加監控和日誌分析的負擔，對管理員造成壓力。

因此，使用 RetryTopic 功能時必須進行適當的管理，以避免對 Apache Kafka 造成不必要的負擔。可以根據需要適當的設置重試次數或間隔，以避免過多的使用 RetryTopic 功能。此外，您可以是當地進行處存和網路寬頻的規劃，以支持隨著 RetryTopic 數量增加而帶來的需求。

其他程式碼及annotation的用法及含意，我都註解寫在code中了， 想知道的話..