GH-2262: Polishing

Resolves #2262

Author: garyrussell

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/BackOffHandler.java

@@ -27,6 +27,12 @@
 @FunctionalInterface
 public interface BackOffHandler {
 
+	/**
+	 * Perform the next back off.
+	 * @param container the container.
+	 * @param exception the exception.
+	 * @param nextBackOff the next back off.
+	 */
 	void onNextBackOff(@Nullable MessageListenerContainer container, Exception exception, long nextBackOff);
 
 }

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerPausingBackOffHandler.java

@@ -0,0 +1,47 @@
+/*
+ * 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 java.time.Duration;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * A {@link BackOffHandler} that pauses the container for the backoff.
+ *
+ * @author Gary Russell
+ * @since 2.9
+ *
+ */
+public class ContainerPausingBackOffHandler implements BackOffHandler {
+
+	private final ListenerContainerPauseService pauser;
+
+	/**
+	 * Create an instance with the provided {@link ListenerContainerPauseService}.
+	 * @param pauser the pause service.
+	 */
+	public ContainerPausingBackOffHandler(ListenerContainerPauseService pauser) {
+		this.pauser = pauser;
+	}
+
+	@Override
+	public void onNextBackOff(@Nullable MessageListenerContainer container, Exception exception, long nextBackOff) {
+		this.pauser.pause(container, Duration.ofMillis(nextBackOff));
+	}
+
+}

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java

@@ -107,29 +107,35 @@ public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>,
 	/**
 	 * Construct an instance with the provided recoverer which will be called after the
 	 * backOff returns STOP for a topic/partition/offset.
-	 * @param recoverer       the recoverer; if null, the default (logging) recoverer is used.
-	 * @param backOff         the {@link BackOff}.
+	 * @param recoverer the recoverer; if null, the default (logging) recoverer is used.
+	 * @param backOff the {@link BackOff}.
 	 * @param kafkaOperations for sending the recovered offset to the transaction.
 	 * @param commitRecovered true to commit the recovered record's offset; requires a
-	 *                        {@link KafkaOperations}.
-	 * @since 2.9
+	 * {@link KafkaOperations}.
+	 * @since 2.5.3
 	 */
-	public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff, @Nullable KafkaOperations<?, ?> kafkaOperations, boolean commitRecovered) {
+	public DefaultAfterRollbackProcessor(@Nullable
+	BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer,
+		BackOff backOff, @Nullable KafkaOperations<?, ?> kafkaOperations, boolean commitRecovered) {
+
 		this(recoverer, backOff, null, kafkaOperations, commitRecovered);
 	}
 
 	/**
 	 * Construct an instance with the provided recoverer which will be called after the
 	 * backOff returns STOP for a topic/partition/offset.
-	 * @param recoverer       the recoverer; if null, the default (logging) recoverer is used.
-	 * @param backOff         the {@link BackOff}.
-	 * @param backOffHandler  the {@link BackOffHandler}.
+	 * @param recoverer the recoverer; if null, the default (logging) recoverer is used.
+	 * @param backOff the {@link BackOff}.
+	 * @param backOffHandler the {@link BackOffHandler}.
 	 * @param kafkaOperations for sending the recovered offset to the transaction.
 	 * @param commitRecovered true to commit the recovered record's offset; requires a
-	 *                        {@link KafkaOperations}.
-	 * @since 2.5.9
+	 * {@link KafkaOperations}.
+	 * @since 2.9
 	 */
-	public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff, @Nullable BackOffHandler backOffHandler, @Nullable KafkaOperations<?, ?> kafkaOperations, boolean commitRecovered) {
+	public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer,
+			BackOff backOff, @Nullable BackOffHandler backOffHandler, @Nullable KafkaOperations<?, ?> kafkaOperations,
+			boolean commitRecovered) {
+
 		super(recoverer, backOff, backOffHandler);
 		this.kafkaTemplate = kafkaOperations;
 		super.setCommitRecovered(commitRecovered);

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java

@@ -23,6 +23,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.common.errors.SerializationException;
 
+import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.backoff.BackOff;
 
@@ -90,13 +91,16 @@ public DefaultErrorHandler(@Nullable ConsumerRecordRecoverer recoverer, BackOff
 	}
 
 	/**
-	 * Construct an instance with the provided recoverer which will be called after
-	 * the backOff returns STOP for a topic/partition/offset.
+	 * Construct an instance with the provided recoverer which will be called after the
+	 * backOff returns STOP for a topic/partition/offset.
 	 * @param recoverer the recoverer; if null, the default (logging) recoverer is used.
 	 * @param backOff the {@link BackOff}.
 	 * @param backOffHandler the {@link BackOffHandler}.
+	 * @since 2.9
 	 */
-	public DefaultErrorHandler(@Nullable ConsumerRecordRecoverer recoverer, BackOff backOff, @Nullable BackOffHandler backOffHandler) {
+	public DefaultErrorHandler(@Nullable ConsumerRecordRecoverer recoverer, BackOff backOff,
+			@Nullable BackOffHandler backOffHandler) {
+
 		super(recoverer, backOff, backOffHandler, createFallback(backOff, recoverer));
 	}
 
@@ -151,6 +155,7 @@ public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record,
 			return getFailureTracker().recovered(record, thrownException, container, consumer);
 		}
 		catch (Exception ex) {
+			logger.error(ex, "Failed to handle " + KafkaUtils.format(record) + " with " + thrownException);
 			return false;
 		}
 	}

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/FailedBatchProcessor.java

@@ -75,6 +75,7 @@ public FailedBatchProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception
 	 * @param backOff the back off.
 	 * @param backOffHandler the {@link BackOffHandler}
 	 * @param fallbackHandler the fall back handler.
+	 * @since 2.9
 	 */
 	public FailedBatchProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff,
 								@Nullable BackOffHandler backOffHandler, CommonErrorHandler fallbackHandler) {

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java

@@ -64,7 +64,9 @@ protected FailedRecordProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Excep
 		this(recoverer, backOff, null);
 	}
 
-	protected FailedRecordProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff, @Nullable BackOffHandler backOffHandler) {
+	protected FailedRecordProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer,
+			BackOff backOff, @Nullable BackOffHandler backOffHandler) {
+
 		this.failureTracker = new FailedRecordTracker(recoverer, backOff, backOffHandler, this.logger);
 		this.failureTracker.setBackOffFunction(this.noRetriesForClassified);
 	}

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordTracker.java

@@ -65,7 +65,13 @@ class FailedRecordTracker implements RecoveryStrategy {
 	private boolean resetStateOnExceptionChange = true;
 
 	FailedRecordTracker(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff,
-						@Nullable BackOffHandler backOffHandler, LogAccessor logger) {
+			LogAccessor logger) {
+
+		this(recoverer, backOff, null, logger);
+	}
+
+	FailedRecordTracker(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff,
+			@Nullable BackOffHandler backOffHandler, LogAccessor logger) {
 
 		Assert.notNull(backOff, "'backOff' cannot be null");
 		if (recoverer == null) {
@@ -77,8 +83,8 @@ class FailedRecordTracker implements RecoveryStrategy {
 				}
 				logger.error(thr, "Backoff "
 						+ (failedRecord == null
-						? "none"
-						: failedRecord.getBackOffExecution())
+								? "none"
+								: failedRecord.getBackOffExecution())
 						+ " exhausted for " + KafkaUtils.format(rec));
 			};
 		}
@@ -97,11 +103,6 @@ class FailedRecordTracker implements RecoveryStrategy {
 
 	}
 
-	FailedRecordTracker(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff,
-						LogAccessor logger) {
-		this(recoverer, backOff, null, logger);
-	}
-
 	/**
 	 * Set a function to dynamically determine the {@link BackOff} to use, based on the
 	 * consumer record and/or exception. If null is returned, the default BackOff will be
@@ -307,6 +308,7 @@ void setLastException(Exception lastException) {
 	}
 
 	static class DefaultBackOffHandler implements BackOffHandler {
+		@Override
 		public void onNextBackOff(@Nullable MessageListenerContainer container, Exception exception, long nextBackOff) {
 			try {
 				if (container == null) {

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerContainerPauseService.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2022 the original author or authors.
+ * 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.
@@ -17,105 +17,93 @@
 package org.springframework.kafka.listener;
 
 import java.time.Duration;
-import java.time.LocalDateTime;
+import java.time.Instant;
+import java.time.ZoneId;
 import java.util.Optional;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.NonNull;
 import org.springframework.lang.Nullable;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.util.Assert;
 
 /**
  * Service for pausing and resuming of {@link MessageListenerContainer}.
  *
  * @author Jan Marincek
+ * @author Gary Russell
  * @since 2.9
  */
 public class ListenerContainerPauseService {
 
 	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(ListenerContainerPauseService.class));
+
+	@Nullable
 	private final ListenerContainerRegistry registry;
-	private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
 
-	public ListenerContainerPauseService(ListenerContainerRegistry registry) {
-		this.registry = registry;
-	}
+	private final TaskScheduler scheduler;
 
 	/**
-	 * Pause the listener by given id.
-	 * Checks if the listener has already been requested to pause.
-	 *
-	 * @param listenerId the id of the listener
+	 * Create an instance with the provided registry and scheduler.
+	 * @param registry the registry or null.
+	 * @param scheduler the scheduler.
 	 */
-	public void pause(String listenerId) {
-		getListenerContainer(listenerId).ifPresent(this::pause);
+	public ListenerContainerPauseService(@Nullable ListenerContainerRegistry registry, TaskScheduler scheduler) {
+		Assert.notNull(scheduler, "'scheduler' cannot be null");
+		this.registry = registry;
+		this.scheduler = scheduler;
 	}
 
 	/**
 	 * Pause the listener by given id.
 	 * Checks if the listener has already been requested to pause.
 	 * Sets executor schedule for resuming the same listener after pauseDuration.
-	 *
 	 * @param listenerId    the id of the listener
 	 * @param pauseDuration duration between pause() and resume() actions
 	 */
 	public void pause(String listenerId, Duration pauseDuration) {
+		Assert.notNull(this.registry, "Pause by id is only supported when a registry is provided");
 		getListenerContainer(listenerId)
 				.ifPresent(messageListenerContainer -> pause(messageListenerContainer, pauseDuration));
 	}
 
 	/**
-	 * Pause the listener by given container instance.
-	 * Checks if the listener has already been requested to pause.
-	 *
+	 * Pause the listener by given container instance. Checks if the listener has already
+	 * been requested to pause. Sets executor schedule for resuming the same listener
+	 * after pauseDuration.
 	 * @param messageListenerContainer the listener container
+	 * @param pauseDuration duration between pause() and resume() actions
 	 */
-	public void pause(@NonNull MessageListenerContainer messageListenerContainer) {
-		pause(messageListenerContainer, null);
-	}
-
-	/**
-	 * Pause the listener by given container instance.
-	 * Checks if the listener has already been requested to pause.
-	 * Sets executor schedule for resuming the same listener after pauseDuration.
-	 *
-	 * @param messageListenerContainer the listener container
-	 * @param pauseDuration            duration between pause() and resume() actions
-	 */
-	public void pause(@NonNull MessageListenerContainer messageListenerContainer, @Nullable Duration pauseDuration) {
+	public void pause(MessageListenerContainer messageListenerContainer, Duration pauseDuration) {
 		if (messageListenerContainer.isPauseRequested()) {
 			LOGGER.debug(() -> "Container " + messageListenerContainer + " already has pause requested");
 		}
 		else {
-			LOGGER.debug(() -> "Pausing container " + messageListenerContainer);
+			Instant resumeAt = Instant.now().plusMillis(pauseDuration.toMillis());
+			LOGGER.debug(() -> "Pausing container " + messageListenerContainer + "resume scheduled for "
+					+ resumeAt.atZone(ZoneId.systemDefault()).toLocalDateTime());
 			messageListenerContainer.pause();
-			if (messageListenerContainer.getListenerId() != null && pauseDuration != null) {
-				LOGGER.debug(() -> "Resuming of container " + messageListenerContainer + " scheduled for " + LocalDateTime.now().plus(pauseDuration));
-				this.executor.schedule(() -> resume(messageListenerContainer.getListenerId()), pauseDuration.toMillis(), TimeUnit.MILLISECONDS);
-			}
+			this.scheduler.schedule(() -> resume(messageListenerContainer), resumeAt);
 		}
 	}
 
 	/**
-	 * Resume the listener by given id.
-	 *
+	 * Resume the listener container by given id.
 	 * @param listenerId the id of the listener
 	 */
 	public void resume(@NonNull String listenerId) {
+		Assert.notNull(this.registry, "Resume by id is only supported when a registry is provided");
 		getListenerContainer(listenerId).ifPresent(this::resume);
 	}
 
 	/**
-	 * Resume the listener.
-	 *
+	 * Resume the listener container.
 	 * @param messageListenerContainer the listener container
 	 */
 	public void resume(@NonNull MessageListenerContainer messageListenerContainer) {
-		if (messageListenerContainer.isContainerPaused()) {
+		if (messageListenerContainer.isPauseRequested()) {
 			LOGGER.debug(() -> "Resuming container " + messageListenerContainer);
 			messageListenerContainer.resume();
 		}