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

GH-1509 Support nested BatchListenerFailedException in RecoveringBatchErrorHandler

Browse Source
This commit is contained in:
mhyeon-lee
2020-06-17 21:57:39 +09:00
committed by Gary Russell
parent 2186774de9
commit 76f8819965
2 changed files with 70 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Throwable> 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;
}
}

41
spring-kafka/src/test/java/org/springframework/kafka/listener/RecoveringBatchErrorHandlerTests.java
View File

@@ -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<TopicPartition, OffsetAndMetadata> 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() {