From 6ec1b3bb02aba18f7795fb9aa90d219f077c1621 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 24 Oct 2022 17:43:46 -0400 Subject: [PATCH] GH-2459: FallbackBatchErrorHandler Retryable Ex Resolves https://github.com/spring-projects/spring-kafka/issues/2459 The `FallbackBatchErrorHandler` was not an `ExceptionClassifier`. The default error handler should propagate exception classifications. **2.8.x**x --- .../kafka/listener/ErrorHandlerAdapter.java | 30 +++++++++- .../kafka/listener/ErrorHandlingUtils.java | 55 ++++++++++++++++++- .../kafka/listener/ExceptionClassifier.java | 12 ++++ .../kafka/listener/FailedBatchProcessor.java | 35 ++++++++++++ .../listener/RetryingBatchErrorHandler.java | 4 +- .../DefaultErrorHandlerBatchTests.java | 28 ++++++++++ 6 files changed, 159 insertions(+), 5 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java index 6f4bd583..82d92150 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java @@ -19,6 +19,8 @@ package org.springframework.kafka.listener; import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.Map; +import java.util.stream.Stream; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -26,6 +28,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.common.TopicPartition; import org.springframework.kafka.support.TopicPartitionOffset; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** @@ -35,7 +38,7 @@ import org.springframework.util.Assert; * @since 2.7.4 * */ -class ErrorHandlerAdapter implements CommonErrorHandler { +class ErrorHandlerAdapter extends ExceptionClassifier implements CommonErrorHandler { @SuppressWarnings({ "rawtypes", "unchecked" }) private static final ConsumerRecords EMPTY_BATCH = new ConsumerRecords(Collections.emptyMap()); @@ -170,5 +173,30 @@ class ErrorHandlerAdapter implements CommonErrorHandler { } } + @Override + protected void notRetryable(Stream> notRetryable) { + if (this.batchErrorHandler instanceof ExceptionClassifier) { + notRetryable.forEach(ex -> ((ExceptionClassifier) this.batchErrorHandler).addNotRetryableExceptions(ex)); + } + } + + @Override + public void setClassifications(Map, Boolean> classifications, boolean defaultValue) { + super.setClassifications(classifications, defaultValue); + if (this.batchErrorHandler instanceof ExceptionClassifier) { + ((ExceptionClassifier) this.batchErrorHandler).setClassifications(classifications, defaultValue); + } + } + + @Override + @Nullable + public Boolean removeClassification(Class exceptionType) { + Boolean removed = super.removeClassification(exceptionType); + if (this.batchErrorHandler instanceof ExceptionClassifier) { + ((ExceptionClassifier) this.batchErrorHandler).removeClassification(exceptionType); + } + return removed; + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java index 8fece665..42d47f57 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java @@ -26,6 +26,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.common.TopicPartition; +import org.springframework.classify.BinaryExceptionClassifier; import org.springframework.core.log.LogAccessor; import org.springframework.kafka.KafkaException; import org.springframework.lang.Nullable; @@ -77,18 +78,49 @@ public final class ErrorHandlingUtils { * @param recoverer the recoverer. * @param logger the logger. * @param logLevel the log level. + * @deprecated in favor of + * {@link #retryBatch(Exception, ConsumerRecords, Consumer, MessageListenerContainer, Runnable, BackOff, CommonErrorHandler, BiConsumer, LogAccessor, org.springframework.kafka.KafkaException.Level, List, BinaryExceptionClassifier)}. */ + @Deprecated public static void retryBatch(Exception thrownException, ConsumerRecords records, Consumer consumer, MessageListenerContainer container, Runnable invokeListener, BackOff backOff, CommonErrorHandler seeker, BiConsumer, Exception> recoverer, LogAccessor logger, KafkaException.Level logLevel) { + retryBatch(thrownException, records, consumer, container, invokeListener, backOff, seeker, null, logger, + logLevel, null, new BinaryExceptionClassifier(true)); + } + + /** + * Retry a complete batch by pausing the consumer and then, in a loop, poll the + * consumer, wait for the next back off, then call the listener. When retries are + * exhausted, call the recoverer with the {@link ConsumerRecords}. + * @param thrownException the exception. + * @param records the records. + * @param consumer the consumer. + * @param container the container. + * @param invokeListener the {@link Runnable} to run (call the listener). + * @param backOff the backOff. + * @param seeker the common error handler that re-seeks the entire batch. + * @param recoverer the recoverer. + * @param logger the logger. + * @param logLevel the log level. + * @param retryListenersArg the retry listeners. + * @param classifier the exception classifier. + * @since 2.8.11 + */ + public static void retryBatch(Exception thrownException, ConsumerRecords records, Consumer consumer, + MessageListenerContainer container, Runnable invokeListener, BackOff backOff, + CommonErrorHandler seeker, BiConsumer, Exception> recoverer, LogAccessor logger, + KafkaException.Level logLevel, @Nullable List retryListenersArg, + BinaryExceptionClassifier classifier) { + BackOffExecution execution = backOff.start(); long nextBackOff = execution.nextBackOff(); String failed = null; Set assignment = consumer.assignment(); consumer.pause(assignment); - List listeners = retryListeners.get(); + List listeners = retryListenersArg != null ? retryListenersArg : retryListeners.get(); int attempt = 1; listen(listeners, records, thrownException, attempt++); ConsumerRecord first = records.iterator().next(); @@ -98,7 +130,8 @@ public final class ErrorHandlingUtils { .publishConsumerPausedEvent(assignment, "For batch retry"); } try { - while (nextBackOff != BackOffExecution.STOP) { + Boolean retryable = classifier.classify(unwrapIfNeeded(thrownException)); + while (Boolean.TRUE.equals(retryable) && nextBackOff != BackOffExecution.STOP) { consumer.poll(Duration.ZERO); try { ListenerUtils.stoppableSleep(container, nextBackOff); @@ -171,4 +204,22 @@ public final class ErrorHandlingUtils { return sb.toString(); } + /** + * Remove a {@link TimestampedException}, if present. + * Remove a {@link ListenerExecutionFailedException}, if present. + * @param exception the exception. + * @return the unwrapped cause or cause of cause. + * @since 2.8.11 + */ + public static Exception unwrapIfNeeded(Exception exception) { + Exception theEx = exception; + if (theEx instanceof TimestampedException && theEx.getCause() instanceof Exception) { + theEx = (Exception) theEx.getCause(); + } + if (theEx instanceof ListenerExecutionFailedException && theEx.getCause() instanceof Exception) { + theEx = (Exception) theEx.getCause(); + } + return theEx; + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java index 2b672a5f..1cdac831 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java @@ -16,8 +16,10 @@ package org.springframework.kafka.listener; +import java.util.Arrays; import java.util.HashMap; import java.util.Map; +import java.util.stream.Stream; import org.springframework.classify.BinaryExceptionClassifier; import org.springframework.kafka.support.converter.ConversionException; @@ -117,6 +119,16 @@ public abstract class ExceptionClassifier extends KafkaExceptionLogLevelAware { @SuppressWarnings("varargs") public final void addNotRetryableExceptions(Class... exceptionTypes) { add(false, exceptionTypes); + notRetryable(Arrays.stream(exceptionTypes)); + } + + /** + * Subclasses can override this to receive notification of configuration of not + * retryable exceptions. + * @param notRetryable the not retryable exceptions. + * @since 2.9.3 + */ + protected void notRetryable(Stream> notRetryable) { } /** diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedBatchProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedBatchProcessor.java index 743d2208..ad848a09 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedBatchProcessor.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedBatchProcessor.java @@ -26,6 +26,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.function.BiConsumer; +import java.util.stream.Stream; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -35,6 +36,7 @@ import org.apache.kafka.clients.consumer.OffsetCommitCallback; import org.apache.kafka.common.TopicPartition; import org.springframework.kafka.KafkaException; +import org.springframework.kafka.KafkaException.Level; import org.springframework.lang.Nullable; import org.springframework.util.backoff.BackOff; @@ -70,6 +72,39 @@ public abstract class FailedBatchProcessor extends FailedRecordProcessor { this.fallbackBatchHandler = fallbackHandler; } + @Override + public void setLogLevel(Level logLevel) { + super.setLogLevel(logLevel); + if (this.fallbackBatchHandler instanceof KafkaExceptionLogLevelAware) { + ((KafkaExceptionLogLevelAware) this.fallbackBatchHandler).setLogLevel(logLevel); + } + } + + @Override + protected void notRetryable(Stream> notRetryable) { + if (this.fallbackBatchHandler instanceof ExceptionClassifier) { + notRetryable.forEach(ex -> ((ExceptionClassifier) this.fallbackBatchHandler).addNotRetryableExceptions(ex)); + } + } + + @Override + public void setClassifications(Map, Boolean> classifications, boolean defaultValue) { + super.setClassifications(classifications, defaultValue); + if (this.fallbackBatchHandler instanceof ExceptionClassifier) { + ((ExceptionClassifier) this.fallbackBatchHandler).setClassifications(classifications, defaultValue); + } + } + + @Override + @Nullable + public Boolean removeClassification(Class exceptionType) { + Boolean removed = super.removeClassification(exceptionType); + if (this.fallbackBatchHandler instanceof ExceptionClassifier) { + ((ExceptionClassifier) this.fallbackBatchHandler).removeClassification(exceptionType); + } + return removed; + } + /** * Return the fallback batch error handler. * @return the handler. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java index 8f991c81..81f61fad 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java @@ -43,7 +43,7 @@ import org.springframework.util.backoff.FixedBackOff; * */ @Deprecated -public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware +public class RetryingBatchErrorHandler extends ExceptionClassifier implements ListenerInvokingBatchErrorHandler { private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); @@ -107,7 +107,7 @@ public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware this.retrying.set(true); try { ErrorHandlingUtils.retryBatch(thrownException, records, consumer, container, invokeListener, this.backOff, - this.seeker, this.recoverer, this.logger, getLogLevel()); + this.seeker, this.recoverer, this.logger, getLogLevel(), null, getClassifier()); } finally { this.retrying.set(false); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultErrorHandlerBatchTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultErrorHandlerBatchTests.java index 96c9eb2c..6b45ca09 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultErrorHandlerBatchTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultErrorHandlerBatchTests.java @@ -24,6 +24,7 @@ import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willAnswer; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -234,6 +235,33 @@ public class DefaultErrorHandlerBatchTests { verify(retryListener).recovered(any(ConsumerRecords.class), any()); } + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Test + void notRetryable() { + Consumer mockConsumer = mock(Consumer.class); + ConsumerRecordRecoverer recoverer = mock(ConsumerRecordRecoverer.class); + DefaultErrorHandler beh = new DefaultErrorHandler(recoverer, new FixedBackOff(0, 2)); + beh.addNotRetryableExceptions(IllegalStateException.class); + RetryListener retryListener = mock(RetryListener.class); + beh.setRetryListeners(retryListener); + TopicPartition tp = new TopicPartition("foo", 0); + ConsumerRecords records = new ConsumerRecords(Collections.singletonMap(tp, + List.of(new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo", + new RecordHeaders(), Optional.empty()), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo", + new RecordHeaders(), Optional.empty())))); + MessageListenerContainer container = mock(MessageListenerContainer.class); + given(container.isRunning()).willReturn(true); + beh.handleBatch(new ListenerExecutionFailedException("test", new IllegalStateException()), + records, mockConsumer, container, () -> { + }); + verify(retryListener).failedDelivery(any(ConsumerRecords.class), any(), eq(1)); + // no retries + verify(retryListener, never()).failedDelivery(any(ConsumerRecords.class), any(), eq(2)); + verify(recoverer, times(2)).accept(any(), any()); // each record in batch + verify(retryListener).recovered(any(ConsumerRecords.class), any()); + } + @Configuration @EnableKafka public static class Config {