Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2459: FallbackBatchErrorHandler Retryable Ex

Browse Source 
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
This commit is contained in:
Gary Russell
2022-10-24 17:43:46 -04:00
parent 48b2017b0d
commit 6ec1b3bb02
6 changed files with 159 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java
View File

@@ -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<Class<? extends Exception>> notRetryable) {
if (this.batchErrorHandler instanceof ExceptionClassifier) {
notRetryable.forEach(ex -> ((ExceptionClassifier) this.batchErrorHandler).addNotRetryableExceptions(ex));
}
}
@Override
public void setClassifications(Map<Class<? extends Throwable>, 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<? extends Exception> exceptionType) {
Boolean removed = super.removeClassification(exceptionType);
if (this.batchErrorHandler instanceof ExceptionClassifier) {
((ExceptionClassifier) this.batchErrorHandler).removeClassification(exceptionType);
}
return removed;
}
}

55
spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java
View File

@@ -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<ConsumerRecords<?, ?>, 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<ConsumerRecords<?, ?>, Exception> recoverer, LogAccessor logger,
KafkaException.Level logLevel, @Nullable List<RetryListener> retryListenersArg,
BinaryExceptionClassifier classifier) {
BackOffExecution execution = backOff.start();
long nextBackOff = execution.nextBackOff();
String failed = null;
Set<TopicPartition> assignment = consumer.assignment();
consumer.pause(assignment);
List<RetryListener> listeners = retryListeners.get();
List<RetryListener> 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;
}
}

12
spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java
View File

@@ -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<? extends Exception>... 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<Class<? extends Exception>> notRetryable) {
}
/**

35
spring-kafka/src/main/java/org/springframework/kafka/listener/FailedBatchProcessor.java
View File

@@ -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<Class<? extends Exception>> notRetryable) {
if (this.fallbackBatchHandler instanceof ExceptionClassifier) {
notRetryable.forEach(ex -> ((ExceptionClassifier) this.fallbackBatchHandler).addNotRetryableExceptions(ex));
}
}
@Override
public void setClassifications(Map<Class<? extends Throwable>, 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<? extends Exception> 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.

4
spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java
View File

@@ -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);

28
spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultErrorHandlerBatchTests.java
View File

@@ -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 {