From b30e4b3a483e0fb3f4febef3b7a85665e648084a Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 9 Feb 2022 12:29:13 -0500 Subject: [PATCH] GH-2099: Deprecate RetryingBatchErrorHandler Resolves https://github.com/spring-projects/spring-kafka/issues/2099 Similar to other legacy error handlers, deprecate this one because its functionality is replaced by the `DefaultErrorHandler`. Eventually move logic to `FallbackBatchErrorHandler` which is package private. **cherry-pick to 2.8.x** --- .../kafka/listener/DefaultErrorHandler.java | 8 +-- .../kafka/listener/FailedBatchProcessor.java | 10 ++-- .../listener/FallbackBatchErrorHandler.java | 53 +++++++++++++++++++ .../listener/RetryingBatchErrorHandler.java | 12 +++-- ...ackBatchErrorHandlerIntegrationTests.java} | 16 +++--- ...va => FallbackBatchErrorHandlerTests.java} | 10 ++-- 6 files changed, 82 insertions(+), 27 deletions(-) create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/FallbackBatchErrorHandler.java rename spring-kafka/src/test/java/org/springframework/kafka/listener/{RetryingBatchErrorHandlerIntegrationTests.java => FallbackBatchErrorHandlerIntegrationTests.java} (95%) rename spring-kafka/src/test/java/org/springframework/kafka/listener/{RetryingBatchErrorHandlerTests.java => FallbackBatchErrorHandlerTests.java} (94%) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java index 152b60cb..ba8e3a62 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2021 the original author or authors. + * Copyright 2021-2022 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. @@ -36,11 +36,11 @@ import org.springframework.util.backoff.BackOff; * the partitions for the remaining records will be repositioned and/or the failed record * can be recovered and skipped. If some other exception is thrown, or a valid record is * not provided in the exception, error handling is delegated to a - * {@link RetryingBatchErrorHandler} with this handler's {@link BackOff}. If the record is + * {@link FallbackBatchErrorHandler} with this handler's {@link BackOff}. If the record is * recovered, its offset is committed. This is a replacement for the legacy * {@link SeekToCurrentErrorHandler} and {@link SeekToCurrentBatchErrorHandler} (but the * fallback now can send the messages to a recoverer after retries are completed instead - * of retring indefinitely). + * of retrying indefinitely). * * @author Gary Russell * @@ -90,7 +90,7 @@ public class DefaultErrorHandler extends FailedBatchProcessor implements CommonE } private static CommonErrorHandler createFallback(BackOff backOff, @Nullable ConsumerRecordRecoverer recoverer) { - return new ErrorHandlerAdapter(new RetryingBatchErrorHandler(backOff, recoverer)); + return new ErrorHandlerAdapter(new FallbackBatchErrorHandler(backOff, recoverer)); } /** 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 e838abac..c219e103 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 @@ -1,5 +1,5 @@ /* - * Copyright 2021 the original author or authors. + * Copyright 2021-2022 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. @@ -54,7 +54,7 @@ public abstract class FailedBatchProcessor extends FailedRecordProcessor { private static final LoggingCommitCallback LOGGING_COMMIT_CALLBACK = new LoggingCommitCallback(); - private final CommonErrorHandler fallbackHandler; + private final CommonErrorHandler fallbackBatchHandler; /** * Construct an instance with the provided properties. @@ -66,7 +66,7 @@ public abstract class FailedBatchProcessor extends FailedRecordProcessor { CommonErrorHandler fallbackHandler) { super(recoverer, backOff); - this.fallbackHandler = fallbackHandler; + this.fallbackBatchHandler = fallbackHandler; } protected void doHandle(Exception thrownException, ConsumerRecords data, Consumer consumer, @@ -75,7 +75,7 @@ public abstract class FailedBatchProcessor extends FailedRecordProcessor { BatchListenerFailedException batchListenerFailedException = getBatchListenerFailedException(thrownException); if (batchListenerFailedException == null) { this.logger.debug(thrownException, "Expected a BatchListenerFailedException; re-seeking batch"); - this.fallbackHandler.handleBatch(thrownException, data, consumer, container, invokeListener); + this.fallbackBatchHandler.handleBatch(thrownException, data, consumer, container, invokeListener); } else { ConsumerRecord record = batchListenerFailedException.getRecord(); @@ -84,7 +84,7 @@ public abstract class FailedBatchProcessor extends FailedRecordProcessor { 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.handleBatch(thrownException, data, consumer, container, invokeListener); + this.fallbackBatchHandler.handleBatch(thrownException, data, consumer, container, invokeListener); } else { seekOrRecover(thrownException, data, consumer, container, index); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/FallbackBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/FallbackBatchErrorHandler.java new file mode 100644 index 00000000..7995f33f --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/FallbackBatchErrorHandler.java @@ -0,0 +1,53 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import org.springframework.lang.Nullable; +import org.springframework.util.backoff.BackOff; +import org.springframework.util.backoff.FixedBackOff; + +/** + * A batch error handler used by the default error handler when the listener does + * not throw a {@link BatchListenerFailedException}. + * + * @author Gary Russell + * @since 2.8.3 + * + */ +@SuppressWarnings("deprecation") +class FallbackBatchErrorHandler extends RetryingBatchErrorHandler { + + /** + * Construct an instance with a default {@link FixedBackOff} (unlimited attempts with + * a 5 second back off). + */ + FallbackBatchErrorHandler() { + super(); + } + + /** + * Construct an instance with the provided {@link BackOff} and + * {@link ConsumerRecordRecoverer}. If the recoverer is {@code null}, the discarded + * records (topic-partition{@literal @}offset) will be logged. + * @param backOff the back off. + * @param recoverer the recoverer. + */ + FallbackBatchErrorHandler(BackOff backOff, @Nullable ConsumerRecordRecoverer recoverer) { + super(backOff, recoverer); + } + +} 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 23940274..5059b0a7 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 @@ -1,5 +1,5 @@ /* - * Copyright 2020-2021 the original author or authors. + * Copyright 2020-2022 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. @@ -37,12 +37,14 @@ import org.springframework.util.backoff.FixedBackOff; * * @author Gary Russell * @since 2.3.7 + * @deprecated in favor of {@link DefaultErrorHandler}. * */ +@Deprecated public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware implements ListenerInvokingBatchErrorHandler { - private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(RetryingBatchErrorHandler.class)); + private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); private final BackOff backOff; @@ -72,7 +74,7 @@ public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware this.backOff = backOff; this.recoverer = (crs, ex) -> { if (recoverer == null) { - LOGGER.error(ex, () -> "Records discarded: " + ErrorHandlingUtils.recordsToString(crs)); + this.logger.error(ex, () -> "Records discarded: " + ErrorHandlingUtils.recordsToString(crs)); } else { crs.spliterator().forEachRemaining(rec -> recoverer.accept(rec, ex)); @@ -95,11 +97,11 @@ public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware Consumer consumer, MessageListenerContainer container, Runnable invokeListener) { if (records == null || records.count() == 0) { - LOGGER.error(thrownException, "Called with no records; consumer exception"); + this.logger.error(thrownException, "Called with no records; consumer exception"); return; } ErrorHandlingUtils.retryBatch(thrownException, records, consumer, container, invokeListener, this.backOff, - this.seeker, this.recoverer, LOGGER, getLogLevel()); + this.seeker, this.recoverer, this.logger, getLogLevel()); } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/RetryingBatchErrorHandlerIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/FallbackBatchErrorHandlerIntegrationTests.java similarity index 95% rename from spring-kafka/src/test/java/org/springframework/kafka/listener/RetryingBatchErrorHandlerIntegrationTests.java rename to spring-kafka/src/test/java/org/springframework/kafka/listener/FallbackBatchErrorHandlerIntegrationTests.java index d16b02c0..fd522831 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/RetryingBatchErrorHandlerIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/FallbackBatchErrorHandlerIntegrationTests.java @@ -56,11 +56,11 @@ import org.springframework.util.backoff.FixedBackOff; * */ @EmbeddedKafka(topics = { - RetryingBatchErrorHandlerIntegrationTests.topic1, - RetryingBatchErrorHandlerIntegrationTests.topic1DLT, - RetryingBatchErrorHandlerIntegrationTests.topic2, - RetryingBatchErrorHandlerIntegrationTests.topic2DLT}) -public class RetryingBatchErrorHandlerIntegrationTests { + FallbackBatchErrorHandlerIntegrationTests.topic1, + FallbackBatchErrorHandlerIntegrationTests.topic1DLT, + FallbackBatchErrorHandlerIntegrationTests.topic2, + FallbackBatchErrorHandlerIntegrationTests.topic2DLT}) +public class FallbackBatchErrorHandlerIntegrationTests { public static final String topic1 = "retryTopic1"; @@ -115,7 +115,7 @@ public class RetryingBatchErrorHandlerIntegrationTests { } }; - RetryingBatchErrorHandler errorHandler = new RetryingBatchErrorHandler(new FixedBackOff(0L, 3), recoverer); + FallbackBatchErrorHandler errorHandler = new FallbackBatchErrorHandler(new FixedBackOff(0L, 3), recoverer); container.setBatchErrorHandler(errorHandler); final CountDownLatch stopLatch = new CountDownLatch(1); container.setApplicationEventPublisher(e -> { @@ -186,7 +186,7 @@ public class RetryingBatchErrorHandlerIntegrationTests { } }; - RetryingBatchErrorHandler errorHandler = new RetryingBatchErrorHandler(new FixedBackOff(0L, 3), recoverer); + FallbackBatchErrorHandler errorHandler = new FallbackBatchErrorHandler(new FixedBackOff(0L, 3), recoverer); container.setBatchErrorHandler(errorHandler); final CountDownLatch stopLatch = new CountDownLatch(1); container.setApplicationEventPublisher(e -> { @@ -226,7 +226,7 @@ public class RetryingBatchErrorHandlerIntegrationTests { KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); CountDownLatch called = new CountDownLatch(1); - container.setBatchErrorHandler(new RetryingBatchErrorHandler() { + container.setBatchErrorHandler(new FallbackBatchErrorHandler() { @Override public void handle(Exception thrownException, ConsumerRecords records, Consumer consumer, diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/RetryingBatchErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/FallbackBatchErrorHandlerTests.java similarity index 94% rename from spring-kafka/src/test/java/org/springframework/kafka/listener/RetryingBatchErrorHandlerTests.java rename to spring-kafka/src/test/java/org/springframework/kafka/listener/FallbackBatchErrorHandlerTests.java index a3da02f2..f320f74d 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/RetryingBatchErrorHandlerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/FallbackBatchErrorHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2022 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. @@ -43,7 +43,7 @@ import org.springframework.util.backoff.FixedBackOff; * @since 2.3.7 * */ -public class RetryingBatchErrorHandlerTests { +public class FallbackBatchErrorHandlerTests { private int invoked; @@ -51,7 +51,7 @@ public class RetryingBatchErrorHandlerTests { void recover() { this.invoked = 0; List> recovered = new ArrayList<>(); - RetryingBatchErrorHandler eh = new RetryingBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) -> { + FallbackBatchErrorHandler eh = new FallbackBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) -> { recovered.add(cr); }); Map>> map = new HashMap<>(); @@ -79,7 +79,7 @@ public class RetryingBatchErrorHandlerTests { void successOnRetry() { this.invoked = 0; List> recovered = new ArrayList<>(); - RetryingBatchErrorHandler eh = new RetryingBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) -> { + FallbackBatchErrorHandler eh = new FallbackBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) -> { recovered.add(cr); }); Map>> map = new HashMap<>(); @@ -104,7 +104,7 @@ public class RetryingBatchErrorHandlerTests { void recoveryFails() { this.invoked = 0; List> recovered = new ArrayList<>(); - RetryingBatchErrorHandler eh = new RetryingBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) -> { + FallbackBatchErrorHandler eh = new FallbackBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) -> { recovered.add(cr); throw new RuntimeException("can't recover"); });