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

GH-2970: minor improvement error handler related

Browse Source 
Fixes: #2970 

* minor improvement error handler related
* remove `FailedRecordProcessor.retryListeners`, reuse FailureTracker's retryListeners
* cleanup related to error handlers
This commit is contained in:
Wang Zhiyang
2023-12-23 02:19:43 +08:00
committed by GitHub
parent 4962670bcb
commit 48258dfeae
12 changed files with 23 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -150,13 +150,13 @@ public class DefaultAfterRollbackProcessor<K, V> extends FailedRecordProcessor
"A KafkaOperations is required when 'commitRecovered' is true");
}
@SuppressWarnings({ "unchecked", "rawtypes", "deprecation" })
@SuppressWarnings({ "unchecked", "rawtypes"})
@Override
public void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer,
@Nullable MessageListenerContainer container, Exception exception, boolean recoverable, EOSMode eosMode) {
if (SeekUtils.doSeeks((List) records, consumer, exception, recoverable,
getFailureTracker()::recovered, container, this.logger)
getFailureTracker(), container, this.logger)
&& isCommitRecovered() && this.kafkaTemplate.isTransactional()) {
ConsumerRecord<K, V> skipped = records.get(0);
this.kafkaTemplate.sendOffsetsToTransaction(

2
spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java
View File

@@ -166,7 +166,7 @@ public class DefaultErrorHandler extends FailedBatchProcessor implements CommonE
Consumer<?, ?> consumer, MessageListenerContainer container) {
SeekUtils.seekOrRecover(thrownException, records, consumer, container, isCommitRecovered(), // NOSONAR
getFailureTracker()::recovered, this.logger, getLogLevel());
getFailureTracker(), this.logger, getLogLevel());
}
@Override

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

@@ -94,9 +94,8 @@ public final class ErrorHandlingUtils {
listen(retryListeners, records, thrownException, attempt++);
ConsumerRecord<?, ?> first = records.iterator().next();
MessageListenerContainer childOrSingle = container.getContainerFor(first.topic(), first.partition());
if (childOrSingle instanceof ConsumerPauseResumeEventPublisher) {
((ConsumerPauseResumeEventPublisher) childOrSingle)
.publishConsumerPausedEvent(assignment, "For batch retry");
if (childOrSingle instanceof ConsumerPauseResumeEventPublisher consumerPauseResumeEventPublisher) {
consumerPauseResumeEventPublisher.publishConsumerPausedEvent(assignment, "For batch retry");
}
try {
Exception recoveryException = thrownException;
@@ -165,7 +164,7 @@ public final class ErrorHandlingUtils {
retryListeners.forEach(listener -> listener.recovered(records, finalRecoveryException));
}
catch (Exception ex) {
logger.error(ex, () -> "Recoverer threw an exception; re-seeking batch");
logger.error(ex, "Recoverer threw an exception; re-seeking batch");
retryListeners.forEach(listener -> listener.recoveryFailed(records, thrownException, ex));
seeker.handleBatch(thrownException, records, consumer, container, NO_OP);
}
@@ -173,8 +172,8 @@ public final class ErrorHandlingUtils {
finally {
Set<TopicPartition> assignment2 = consumer.assignment();
consumer.resume(assignment2);
if (childOrSingle instanceof ConsumerPauseResumeEventPublisher) {
((ConsumerPauseResumeEventPublisher) childOrSingle).publishConsumerResumedEvent(assignment2);
if (childOrSingle instanceof ConsumerPauseResumeEventPublisher consumerPauseResumeEventPublisher) {
consumerPauseResumeEventPublisher.publishConsumerResumedEvent(assignment2);
}
}
} // NOSONAR NCSS line count

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

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -219,7 +219,6 @@ public abstract class ExceptionClassifier extends KafkaExceptionLogLevelAware {
* @author Gary Russell
*
*/
@SuppressWarnings("serial")
private static final class ExtendedBinaryExceptionClassifier extends BinaryExceptionClassifier {
ExtendedBinaryExceptionClassifier(Map<Class<? extends Throwable>, Boolean> typeMap, boolean defaultValue) {

10
spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,8 +16,6 @@
package org.springframework.kafka.listener;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
@@ -56,8 +54,6 @@ public abstract class FailedRecordProcessor extends ExceptionClassifier implemen
private final FailedRecordTracker failureTracker;
private final List<RetryListener> retryListeners = new ArrayList<>();
private boolean commitRecovered;
private BiFunction<ConsumerRecord<?, ?>, Exception, BackOff> userBackOffFunction = (rec, ex) -> null;
@@ -136,12 +132,10 @@ public abstract class FailedRecordProcessor extends ExceptionClassifier implemen
public void setRetryListeners(RetryListener... listeners) {
Assert.noNullElements(listeners, "'listeners' cannot have null elements");
this.failureTracker.setRetryListeners(listeners);
this.retryListeners.clear();
this.retryListeners.addAll(Arrays.asList(listeners));
}
protected List<RetryListener> getRetryListeners() {
return this.retryListeners;
return this.failureTracker.getRetryListeners();
}
/**

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

@@ -90,8 +90,8 @@ class FailedRecordTracker implements RecoveryStrategy {
};
}
else {
if (recoverer instanceof ConsumerAwareRecordRecoverer) {
this.recoverer = (ConsumerAwareRecordRecoverer) recoverer;
if (recoverer instanceof ConsumerAwareRecordRecoverer carr) {
this.recoverer = carr;
}
else {
this.recoverer = (rec, consumer, ex) -> recoverer.accept(rec, ex);

2
spring-kafka/src/main/java/org/springframework/kafka/listener/FallbackBatchErrorHandler.java
View File

@@ -181,7 +181,7 @@ class FallbackBatchErrorHandler extends ExceptionClassifier implements CommonErr
.stream()
.collect(
Collectors.toMap(tp -> tp,
tp -> data.records(tp).get(0).offset(), (u, v) -> (long) v, LinkedHashMap::new))
tp -> data.records(tp).get(0).offset(), (u, v) -> v, LinkedHashMap::new))
.forEach(consumer::seek);
throw new KafkaException("Seek to current after exception", getLogLevel(), thrownException);

7
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -1458,9 +1458,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
ConsumerRecords<K, V> pending = this.remainingRecords;
this.remainingRecords = null;
List<ConsumerRecord<?, ?>> records = new ArrayList<>();
Iterator<ConsumerRecord<K, V>> iterator = pending.iterator();
while (iterator.hasNext()) {
records.add(iterator.next());
for (ConsumerRecord<K, V> kvConsumerRecord : pending) {
records.add(kvConsumerRecord);
}
this.commonErrorHandler.handleRemaining(cfe, records, this.consumer,
KafkaMessageListenerContainer.this.thisOrParentContainer);
@@ -2403,7 +2402,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
ConsumerRecords<K, V> records = recordsArg;
List<ConsumerRecord<K, V>> recordList = recordListArg;
if (this.listenerinfo != null) {
records.iterator().forEachRemaining(rec -> listenerInfo(rec));
records.iterator().forEachRemaining(this::listenerInfo);
}
if (this.batchInterceptor != null) {
records = this.batchInterceptor.intercept(recordsArg, this.consumer);

2
spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java
View File

@@ -135,7 +135,7 @@ public final class ListenerUtils {
if (interval == BackOffExecution.STOP) {
interval = lastIntervals.get(currentThread);
if (interval == null) {
interval = Long.valueOf(0);
interval = 0L;
}
}
lastIntervals.put(currentThread, interval);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2022 the original author or authors.
* Copyright 2018-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -204,7 +204,7 @@ public final class SeekUtils {
}
}
if (records == null || !doSeeks(records, consumer, thrownException, true, recovery, container, logger)) { // NOSONAR
if (!doSeeks(records, consumer, thrownException, true, recovery, container, logger)) { // NOSONAR
throw new KafkaException("Seek to current after exception", level, thrownException);
}
if (commitRecovered) {

8
spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultErrorHandlerRecordTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -93,8 +93,6 @@ public class DefaultErrorHandlerRecordTests {
});
ConsumerRecord<String, String> record1 = new ConsumerRecord<>("foo", 0, 0L, "foo", "bar");
ConsumerRecord<String, String> record2 = new ConsumerRecord<>("foo", 1, 1L, "foo", "bar");
List<ConsumerRecord<?, ?>> records = Arrays.asList(record1, record2);
IllegalStateException illegalState = new IllegalStateException();
Consumer<?, ?> consumer = mock(Consumer.class);
assertThat(handler.handleOne(illegalState, record1, consumer, mock(MessageListenerContainer.class))).isFalse();
@@ -116,7 +114,7 @@ public class DefaultErrorHandlerRecordTests {
assertThat(failedDeliveryAttempt.get()).isEqualTo(1);
assertThat(recoveryFailureEx.get())
.isInstanceOf(RuntimeException.class)
.extracting(ex -> ex.getMessage())
.extracting(Throwable::getMessage)
.isEqualTo("test recoverer failure");
assertThat(isRecovered.get()).isTrue();
}
@@ -183,7 +181,7 @@ public class DefaultErrorHandlerRecordTests {
assertThat(failedDeliveryAttempt.get()).isEqualTo(1);
assertThat(recoveryFailureEx.get())
.isInstanceOf(RuntimeException.class)
.extracting(ex -> ex.getMessage())
.extracting(Throwable::getMessage)
.isEqualTo("test recoverer failure");
assertThat(isRecovered.get()).isTrue();
}

16
spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
View File

@@ -279,13 +279,6 @@ public class KafkaMessageListenerContainerTests {
.isEqualTo(ListenerType.SIMPLE);
template.sendDefault(0, 0, "foo");
assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue();
// verify that the container called the right method - avoiding the creation of an Acknowledgment
// assertThat(trace.get()[1].getMethodName()).contains("onMessage"); // onMessage(d, a, c) (inner)
// assertThat(trace.get()[2].getMethodName()).contains("onMessage"); // bridge
// assertThat(trace.get()[3].getMethodName()).contains("onMessage"); // onMessage(d, a, c) (outer)
// assertThat(trace.get()[4].getMethodName()).contains("onMessage"); // onMessage(d)
// assertThat(trace.get()[5].getMethodName()).contains("onMessage"); // bridge
// assertThat(trace.get()[6].getMethodName()).contains("invokeRecordListener");
container.stop();
final CountDownLatch latch3 = new CountDownLatch(1);
filtering = new FilteringMessageListenerAdapter<>(
@@ -299,15 +292,6 @@ public class KafkaMessageListenerContainerTests {
.isEqualTo(ListenerType.ACKNOWLEDGING_CONSUMER_AWARE);
template.sendDefault(0, 0, "foo");
assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue();
// verify that the container called the 3 arg method directly
// int i = 0;
// if (trace.get()[1].getClassName().endsWith("AcknowledgingConsumerAwareMessageListener")) {
// // this frame does not appear in eclise, but does in gradle.\
// i++;
// }
// assertThat(trace.get()[i + 1].getMethodName()).contains("onMessage"); // onMessage(d, a, c)
// assertThat(trace.get()[i + 2].getMethodName()).contains("onMessage"); // bridge
// assertThat(trace.get()[i + 3].getMethodName()).contains("invokeRecordListener");
container.stop();
long t = System.currentTimeMillis();
container.stop();