SpringKafka Consumer DefaultErrorHandler Sınıfı

Giriş

DefaultErrorHandler kullanırken acknowledgement mode ne olmalı emin değilim. Sanırım auto kullanılsa da olur. Buradaki örnekte acknowledgement mode manual yapılıyor gerçi. Açıklaması şöyle

The default behavior is attempting to consume one message at most 10 times, then consume the next message and print an error log if it still fails.

Çıktısı şöyle

2023-06-03T08:57:16.573Z ERROR [order-query-side,,] 1 --- 
[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] 
o.s.kafka.listener.DefaultErrorHandler   : Backoff FixedBackOff
{interval=0, currentAttempts=10, maxAttempts=9} exhausted for ORDER-0@0

Açıklaması şöyle

Blocking Retry

Before we send the fail-processed message to the retry topic, we might want to retry a couple of times to save some network round trip. There are plenty of ways to change the default behavior likes:

1. provide your own @Bean of KafkaListenerErrorHandler

2. provide your own @Bean of DefaultErrorHandler

with different ConsumerRecordRecoverer (instead of just printing error logs) and different BackOff settings to customize attempts and retry intervals.

addNotRetryableExceptions metodu

Örnek

Şöyle yaparız

@Configuration
@Slf4j
public class KafkaConfiguration {
  @Bean
  public DefaultErrorHandler errorHandler() {
    BackOff fixedBackOff = new FixedBackOff(5000, 3);
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(
      (consumerRecord, exception) -> {
      log.error("Couldn't process message: {}; {}", 
        consumerRecord.value().toString(), 
        exception.toString());
    }, fixedBackOff);

    errorHandler.addNotRetryableExceptions(NullPointerException.class);
    return errorHandler;
  }
}

Açıklaması şöyle

Here we have specified that in case of an error, we will do retries (maximum three times) at intervals of five seconds. But if we have an NPE, we won't do iterations in that case but just write a message to the log and skip the message.