From 76f8819965eb9974a1bbe3e3a1ca1a74283394a2 Mon Sep 17 00:00:00 2001 From: mhyeon-lee Date: Wed, 17 Jun 2020 21:57:39 +0900 Subject: [PATCH] GH-1509 Support nested BatchListenerFailedException in RecoveringBatchErrorHandler --- .../listener/RecoveringBatchErrorHandler.java | 35 +++++++++++++--- .../RecoveringBatchErrorHandlerTests.java | 41 +++++++++++++++++++ 2 files changed, 70 insertions(+), 6 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/RecoveringBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/RecoveringBatchErrorHandler.java index 005cdda1..604f6715 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/RecoveringBatchErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/RecoveringBatchErrorHandler.java @@ -20,9 +20,11 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.function.BiConsumer; import org.apache.kafka.clients.consumer.Consumer; @@ -47,6 +49,7 @@ import org.springframework.util.backoff.BackOff; * this handler's {@link BackOff}. If the record is recovered, its offset is committed. * * @author Gary Russell + * @author Myeonghyeon Lee * @since 2.5 * */ @@ -103,16 +106,16 @@ public class RecoveringBatchErrorHandler extends FailedRecordProcessor public void handle(Exception thrownException, ConsumerRecords data, Consumer consumer, MessageListenerContainer container) { - Throwable cause = thrownException.getCause(); - if (!(cause instanceof BatchListenerFailedException)) { - this.logger.debug(cause, "Expected a BatchListenerFailedException; re-seeking batch"); + BatchListenerFailedException batchListenerFailedException = getBatchListenerFailedException(thrownException); + if (batchListenerFailedException == null) { + this.logger.debug(thrownException, "Expected a BatchListenerFailedException; re-seeking batch"); this.fallbackHandler.handle(thrownException, data, consumer, container); } else { - ConsumerRecord record = ((BatchListenerFailedException) cause).getRecord(); - int index = record != null ? findIndex(data, record) : ((BatchListenerFailedException) cause).getIndex(); + ConsumerRecord record = batchListenerFailedException.getRecord(); + int index = record != null ? findIndex(data, record) : batchListenerFailedException.getIndex(); if (index < 0 || index >= data.count()) { - this.logger.warn(cause, () -> String.format("Record not found in batch: %s-%d@%d; re-seeking batch", + this.logger.warn(batchListenerFailedException, () -> String.format("Record not found in batch: %s-%d@%d; re-seeking batch", record.topic(), record.partition(), record.offset())); this.fallbackHandler.handle(thrownException, data, consumer, container); } @@ -191,4 +194,24 @@ public class RecoveringBatchErrorHandler extends FailedRecordProcessor } } + private BatchListenerFailedException getBatchListenerFailedException(Throwable throwable) { + if (throwable == null || throwable instanceof BatchListenerFailedException) { + return (BatchListenerFailedException) throwable; + } + + BatchListenerFailedException target = null; + + Set checked = new HashSet<>(); + while (throwable.getCause() != null && !checked.contains(throwable.getCause())) { + throwable = throwable.getCause(); + checked.add(throwable); + + if (throwable instanceof BatchListenerFailedException) { + target = (BatchListenerFailedException) throwable; + break; + } + } + + return target; + } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/RecoveringBatchErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/RecoveringBatchErrorHandlerTests.java index 18f9151e..6de42ee2 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/RecoveringBatchErrorHandlerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/RecoveringBatchErrorHandlerTests.java @@ -27,6 +27,7 @@ import static org.mockito.Mockito.verify; import java.time.Duration; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.LinkedHashMap; @@ -56,12 +57,14 @@ import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.config.KafkaListenerEndpointRegistry; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.MessagingException; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.util.backoff.FixedBackOff; /** * @author Gary Russell + * @author Myeonghyeon Lee * @since 2.5 * */ @@ -129,6 +132,44 @@ public class RecoveringBatchErrorHandlerTests { verify(mockConsumer).seek(tp, 0L); } + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Test + void wrappedBatchListenerFailedException() { + Consumer mockConsumer = mock(Consumer.class); + InOrder inOrder = inOrder(mockConsumer); + + MessageListenerContainer container = mock(MessageListenerContainer.class); + ContainerProperties containerProperties = mock(ContainerProperties.class); + given(container.getContainerProperties()).willReturn(containerProperties); + given(containerProperties.isSyncCommits()).willReturn(true); + + Duration syncCommitTimeout = Duration.ofMillis(1000); + given(containerProperties.getSyncCommitTimeout()).willReturn(syncCommitTimeout); + + RecoveringBatchErrorHandler beh = new RecoveringBatchErrorHandler(new FixedBackOff(0, 0)); + TopicPartition tp = new TopicPartition("foo", 0); + ConsumerRecords records = new ConsumerRecords(Collections.singletonMap(tp, + Arrays.asList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"), + new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz")) + )); + assertThatExceptionOfType(KafkaException.class).isThrownBy(() -> + beh.handle(new ListenerExecutionFailedException("", new MessagingException("", + new BatchListenerFailedException("", 1))), records, mockConsumer, container) + ); + + Map offsets = new LinkedHashMap<>(); + offsets.put(tp, new OffsetAndMetadata(1L)); + inOrder.verify(mockConsumer).commitSync(offsets, syncCommitTimeout); + + inOrder.verify(mockConsumer).seek(tp, 2); + + offsets = new LinkedHashMap<>(); + offsets.put(tp, new OffsetAndMetadata(2L)); + inOrder.verify(mockConsumer).commitSync(offsets, syncCommitTimeout); + } + @SuppressWarnings({ "unchecked", "rawtypes" }) @Test void missingRecord() {