GH-1744: Improve KafkaConsumerBackoffManager dependency management

Resolves #1744

Introduces the KafkaConsumerTimingAdjuster interface
Introduces the KafkaBackOffManagerFactory interface

Polish code as per code review

Polishing.

spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractKafkaBackOffManagerFactory.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2018-2021 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.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.kafka.config.KafkaListenerConfigUtils;
+import org.springframework.util.Assert;
+
+/**
+ * Base class for {@link KafkaBackOffManagerFactory} implementations.
+ *
+ * @author Tomaz Fernandes
+ * @since 2.7
+ * @see KafkaConsumerBackoffManager
+ */
+public abstract class AbstractKafkaBackOffManagerFactory
+		implements KafkaBackOffManagerFactory, ApplicationContextAware {
+
+	private ApplicationContext applicationContext;
+	private ListenerContainerRegistry listenerContainerRegistry;
+
+	/**
+	 * Creates an instance with the provided {@link ListenerContainerRegistry},
+	 * which will be used to fetch the {@link MessageListenerContainer} to back off.
+
+	 * @param listenerContainerRegistry the listenerContainerRegistry to use.
+	 */
+	public AbstractKafkaBackOffManagerFactory(ListenerContainerRegistry listenerContainerRegistry) {
+		this.listenerContainerRegistry = listenerContainerRegistry;
+	}
+
+	/**
+	 * Creates an instance that will retrieve the {@link ListenerContainerRegistry} from
+	 * the {@link ApplicationContext}.
+	 */
+	public AbstractKafkaBackOffManagerFactory() {
+		this.listenerContainerRegistry = null;
+	}
+
+	/**
+	 * Sets the {@link ListenerContainerRegistry}, that will be used to fetch the
+	 * {@link MessageListenerContainer} to back off.
+	 *
+	 * @param listenerContainerRegistry the listenerContainerRegistry to use.
+	 */
+	public void setListenerContainerRegistry(ListenerContainerRegistry listenerContainerRegistry) {
+		this.listenerContainerRegistry = listenerContainerRegistry;
+	}
+
+	@Override
+	public KafkaConsumerBackoffManager create() {
+		return doCreateManager(getListenerContainerRegistry());
+	}
+
+	protected abstract KafkaConsumerBackoffManager doCreateManager(ListenerContainerRegistry registry);
+
+	protected ListenerContainerRegistry getListenerContainerRegistry() {
+		return this.listenerContainerRegistry != null
+				? this.listenerContainerRegistry
+				: getListenerContainerFromContext();
+	}
+
+	private ListenerContainerRegistry getListenerContainerFromContext() {
+		Assert.notNull(this.applicationContext, "ApplicationContext not set.");
+		return this.applicationContext.getBean(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
+				ListenerContainerRegistry.class);
+	}
+
+	protected <T> T getBean(String beanName, Class<T> beanClass) {
+		return this.applicationContext.getBean(beanName, beanClass);
+	}
+
+	protected void addApplicationListener(ApplicationListener<?> applicationListener) {
+		((ConfigurableApplicationContext) this.applicationContext).addApplicationListener(applicationListener);
+	}
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) {
+		this.applicationContext = applicationContext;
+	}
+}

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaBackOffManagerFactory.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018-2021 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;
+
+/**
+ *
+ * Creates a {@link KafkaBackOffManagerFactory} instance.
+ *
+ * @author Tomaz Fernandes
+ * @since 2.7
+ * @see KafkaConsumerBackoffManager
+ */
+public interface KafkaBackOffManagerFactory {
+
+	KafkaConsumerBackoffManager create();
+
+}

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerBackoffManager.java

@@ -25,13 +25,10 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
 
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationListener;
 import org.springframework.core.log.LogAccessor;
-import org.springframework.core.task.TaskExecutor;
 import org.springframework.kafka.event.ListenerContainerPartitionIdleEvent;
 import org.springframework.lang.Nullable;
-import org.springframework.retry.backoff.Sleeper;
 
 /**
  *
@@ -42,8 +39,7 @@ import org.springframework.retry.backoff.Sleeper;
  * so the Manager can resume the partition consumption.
  *
  * Note that when a record backs off the partition consumption gets paused for
- * approximately that amount of time, so you must have a fixed backoff value per partition
- * in order to make sure no record waits more than it should.
+ * approximately that amount of time, so you must have a fixed backoff value per partition.
  *
  * @author Tomaz Fernandes
  * @author Gary Russell
@@ -53,34 +49,85 @@ import org.springframework.retry.backoff.Sleeper;
 public class KafkaConsumerBackoffManager implements ApplicationListener<ListenerContainerPartitionIdleEvent> {
 
 	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(KafkaConsumerBackoffManager.class));
-	/**
-	 * Internal Back Off Clock Bean Name.
-	 */
-	public static final String INTERNAL_BACKOFF_CLOCK_BEAN_NAME = "internalBackOffClock";
 
-	private static final int TIMING_CORRECTION_THRESHOLD = 100;
-
-	private static final int POLL_TIMEOUTS_FOR_CORRECTION_WINDOW = 2;
-
-	private final ListenerContainerRegistry registry;
+	private final ListenerContainerRegistry listenerContainerRegistry;
 
 	private final Map<TopicPartition, Context> backOffContexts;
 
 	private final Clock clock;
 
-	private final TaskExecutor taskExecutor;
+	private final KafkaConsumerTimingAdjuster kafkaConsumerTimingAdjuster;
 
-	private final Sleeper sleeper;
+	/**
+	 * Constructs an instance with the provided {@link ListenerContainerRegistry} and
+	 * {@link KafkaConsumerTimingAdjuster}.
+	 *
+	 * The ListenerContainerRegistry is used to fetch the {@link MessageListenerContainer}
+	 * that will be backed off / resumed.
+	 *
+	 * The KafkaConsumerTimingAdjuster is used to make timing adjustments
+	 * in the message consumption so that it processes the message closer
+	 * to its due time rather than later.
+	 *
+	 * @param listenerContainerRegistry the listenerContainerRegistry to use.
+	 * @param kafkaConsumerTimingAdjuster the kafkaConsumerTimingAdjuster to use.
+	 */
+	public KafkaConsumerBackoffManager(ListenerContainerRegistry listenerContainerRegistry,
+									KafkaConsumerTimingAdjuster kafkaConsumerTimingAdjuster) {
 
-	public KafkaConsumerBackoffManager(ListenerContainerRegistry registry,
-									@Qualifier(INTERNAL_BACKOFF_CLOCK_BEAN_NAME) Clock clock,
-									TaskExecutor taskExecutor,
-									Sleeper sleeper) {
+		this.listenerContainerRegistry = listenerContainerRegistry;
+		this.kafkaConsumerTimingAdjuster = kafkaConsumerTimingAdjuster;
+		this.clock = Clock.systemUTC();
+		this.backOffContexts = new HashMap<>();
+	}
 
-		this.registry = registry;
+	/**
+	 * Constructs an instance with the provided {@link ListenerContainerRegistry}
+	 * and with no timing adjustment capabilities.
+	 *
+	 * The ListenerContainerRegistry is used to fetch the {@link MessageListenerContainer}
+	 * that will be backed off / resumed.
+	 *
+	 * @param listenerContainerRegistry the listenerContainerRegistry to use.
+	 */
+	public KafkaConsumerBackoffManager(ListenerContainerRegistry listenerContainerRegistry) {
+
+		this.listenerContainerRegistry = listenerContainerRegistry;
+		this.kafkaConsumerTimingAdjuster = null;
+		this.clock = Clock.systemUTC();
+		this.backOffContexts = new HashMap<>();
+	}
+
+	/**
+	 * Creates an instance with the provided {@link ListenerContainerRegistry},
+	 * {@link KafkaConsumerTimingAdjuster} and {@link Clock}.
+	 *
+	 * @param listenerContainerRegistry the listenerContainerRegistry to use.
+	 * @param kafkaConsumerTimingAdjuster the kafkaConsumerTimingAdjuster to use.
+	 * @param clock the clock to use.
+	 */
+	public KafkaConsumerBackoffManager(ListenerContainerRegistry listenerContainerRegistry,
+									KafkaConsumerTimingAdjuster kafkaConsumerTimingAdjuster,
+									Clock clock) {
+
+		this.listenerContainerRegistry = listenerContainerRegistry;
 		this.clock = clock;
-		this.taskExecutor = taskExecutor;
-		this.sleeper = sleeper;
+		this.kafkaConsumerTimingAdjuster = kafkaConsumerTimingAdjuster;
+		this.backOffContexts = new HashMap<>();
+	}
+
+	/**
+	 * Creates an instance with the provided {@link ListenerContainerRegistry}
+	 * and {@link Clock}, with no timing adjustment capabilities.
+	 *
+	 * @param listenerContainerRegistry the listenerContainerRegistry to use.
+	 * @param clock the clock to use.
+	 */
+	public KafkaConsumerBackoffManager(ListenerContainerRegistry listenerContainerRegistry, Clock clock) {
+
+		this.listenerContainerRegistry = listenerContainerRegistry;
+		this.clock = clock;
+		this.kafkaConsumerTimingAdjuster = null;
 		this.backOffContexts = new HashMap<>();
 	}
 
@@ -112,10 +159,6 @@ public class KafkaConsumerBackoffManager implements ApplicationListener<Listener
 				getCurrentMillisFromClock(), partitionIdleEvent.getTopicPartition()));
 
 		Context backOffContext = getBackOffContext(partitionIdleEvent.getTopicPartition());
-
-		if (backOffContext == null) {
-			return;
-		}
 		maybeResumeConsumption(backOffContext);
 	}
 
@@ -123,7 +166,10 @@ public class KafkaConsumerBackoffManager implements ApplicationListener<Listener
 		return Instant.now(this.clock).toEpochMilli();
 	}
 
-	private void maybeResumeConsumption(Context context) {
+	private void maybeResumeConsumption(@Nullable Context context) {
+		if (context == null) {
+			return;
+		}
 		long now = getCurrentMillisFromClock();
 		long timeUntilDue = context.dueTimestamp - now;
 		long pollTimeout = getListenerContainerFromContext(context)
@@ -131,7 +177,9 @@ public class KafkaConsumerBackoffManager implements ApplicationListener<Listener
 				.getPollTimeout();
 		boolean isDue = timeUntilDue <= pollTimeout;
 
-		if (maybeApplyTimingCorrection(context, pollTimeout, timeUntilDue) || isDue) {
+		long adjustedAmount = applyTimingAdjustment(context, timeUntilDue, pollTimeout);
+
+		if (adjustedAmount != 0L || isDue) {
 			resumePartition(context);
 		}
 		else {
@@ -140,6 +188,16 @@ public class KafkaConsumerBackoffManager implements ApplicationListener<Listener
 		}
 	}
 
+	private long applyTimingAdjustment(Context context, long timeUntilDue, long pollTimeout) {
+		if (this.kafkaConsumerTimingAdjuster == null || context.consumerForTimingAdjustment == null) {
+			LOGGER.debug(() -> String.format(
+					"Skipping timing adjustment for TopicPartition %s.", context.topicPartition));
+			return 0L;
+		}
+		return this.kafkaConsumerTimingAdjuster.adjustTiming(
+						context.consumerForTimingAdjustment, context.topicPartition, pollTimeout, timeUntilDue);
+	}
+
 	private void resumePartition(Context context) {
 		MessageListenerContainer container = getListenerContainerFromContext(context);
 		LOGGER.debug(() -> "Resuming partition at " + getCurrentMillisFromClock());
@@ -147,45 +205,8 @@ public class KafkaConsumerBackoffManager implements ApplicationListener<Listener
 		removeBackoff(context.topicPartition);
 	}
 
-	private boolean maybeApplyTimingCorrection(Context context, long pollTimeout, long timeUntilDue) {
-		// Correction can only be applied to ConsumerAwareMessageListener
-		// listener instances.
-		if (context.consumerForTimingCorrection == null) {
-			return false;
-		}
-
-		boolean isInCorrectionWindow = timeUntilDue > pollTimeout && timeUntilDue <=
-				pollTimeout * POLL_TIMEOUTS_FOR_CORRECTION_WINDOW;
-
-		long correctionAmount = timeUntilDue % pollTimeout;
-		if (isInCorrectionWindow && correctionAmount > TIMING_CORRECTION_THRESHOLD) {
-			this.taskExecutor.execute(() -> doApplyTimingCorrection(context, correctionAmount));
-			return true;
-		}
-		return false;
-	}
-
-	private void doApplyTimingCorrection(Context context, long correctionAmount) {
-		try {
-			LOGGER.debug(() -> String.format("Applying correction of %s millis at %s for TopicPartition %s",
-					correctionAmount, getCurrentMillisFromClock(), context.topicPartition));
-			this.sleeper.sleep(correctionAmount);
-			LOGGER.debug(() -> "Waking up consumer for partition topic: " + context.topicPartition);
-			context.consumerForTimingCorrection.wakeup();
-		}
-		catch (InterruptedException e) {
-			Thread.currentThread().interrupt();
-			throw new IllegalStateException("Interrupted waking up consumer while applying correction " +
-					"for TopicPartition " + context.topicPartition, e);
-		}
-		catch (Exception e) { // NOSONAR
-			LOGGER.error(e, () -> "Error waking up consumer while applying correction " +
-					"for TopicPartition " + context.topicPartition);
-		}
-	}
-
 	private MessageListenerContainer getListenerContainerFromContext(Context context) {
-		return this.registry.getListenerContainer(context.listenerId);
+		return this.listenerContainerRegistry.getListenerContainer(context.listenerId);
 	}
 
 	protected void addBackoff(Context context, TopicPartition topicPartition) {
@@ -194,8 +215,7 @@ public class KafkaConsumerBackoffManager implements ApplicationListener<Listener
 		}
 	}
 
-	@Nullable
-	protected Context getBackOffContext(TopicPartition topicPartition) {
+	protected @Nullable Context getBackOffContext(TopicPartition topicPartition) {
 		synchronized (this.backOffContexts) {
 			return this.backOffContexts.get(topicPartition);
 		}
@@ -208,8 +228,8 @@ public class KafkaConsumerBackoffManager implements ApplicationListener<Listener
 	}
 
 	public Context createContext(long dueTimestamp, String listenerId, TopicPartition topicPartition,
-								@Nullable Consumer<?, ?> consumerForTimingCorrection) {
-		return new Context(dueTimestamp, topicPartition, listenerId, consumerForTimingCorrection);
+								@Nullable Consumer<?, ?> consumerForTimingAdjustment) {
+		return new Context(dueTimestamp, topicPartition, listenerId, consumerForTimingAdjustment);
 	}
 
 	/**
@@ -237,14 +257,14 @@ public class KafkaConsumerBackoffManager implements ApplicationListener<Listener
 		/**
 		 * The consumer of the message, if present.
 		 */
-		private final Consumer<?, ?> consumerForTimingCorrection; // NOSONAR
+		private final Consumer<?, ?> consumerForTimingAdjustment; // NOSONAR
 
 		Context(long dueTimestamp, TopicPartition topicPartition, String listenerId,
-						@Nullable Consumer<?, ?> consumerForTimingCorrection) {
+						@Nullable Consumer<?, ?> consumerForTimingAdjustment) {
 			this.dueTimestamp = dueTimestamp;
 			this.listenerId = listenerId;
 			this.topicPartition = topicPartition;
-			this.consumerForTimingCorrection = consumerForTimingCorrection;
+			this.consumerForTimingAdjustment = consumerForTimingAdjustment;
 		}
 	}
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerTimingAdjuster.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2018-2021 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.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.TopicPartition;
+
+/**
+ *
+ * Adjusts the consumption timing of the given consumer to try to have it consume the
+ * next message at a given time until due. Since the {@link org.apache.kafka.clients.consumer.KafkaConsumer}
+ * executes on a single thread, this is done in a best-effort basis.
+ *
+ * @author Tomaz Fernandes
+ * @since 2.7
+ * @see KafkaConsumerBackoffManager
+ */
+public interface KafkaConsumerTimingAdjuster {
+
+	/**
+	 * Executes the timing adjustment.
+	 *
+	 * @param consumerToAdjust the consumer that will have consumption adjusted
+	 * @param topicPartitionToAdjust the consumer's topic partition to be adjusted
+	 * @param containerPollTimeout the consumer's container pollTimeout property
+	 * @param timeUntilNextMessageIsDue the time when the next message should be consumed
+	 *
+	 * @return the applied adjustment amount
+	 */
+	long adjustTiming(Consumer<?, ?> consumerToAdjust, TopicPartition topicPartitionToAdjust,
+							long containerPollTimeout, long timeUntilNextMessageIsDue);
+}

spring-kafka/src/main/java/org/springframework/kafka/listener/TimingAdjustingKafkaBackOffManagerFactory.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2018-2021 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.Clock;
+
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ContextClosedEvent;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.util.Assert;
+
+/**
+ *
+ * Creates a {@link KafkaConsumerBackoffManager} instance
+ * with or without a {@link KafkaConsumerTimingAdjuster}.
+ *
+ * @author Tomaz Fernandes
+ * @since 2.7
+ */
+public class TimingAdjustingKafkaBackOffManagerFactory extends AbstractKafkaBackOffManagerFactory {
+
+	private boolean timingAdjustmentEnabled = true;
+
+	private KafkaConsumerTimingAdjuster timingAdjustmentManager;
+
+	private TaskExecutor taskExecutor;
+
+	private final Clock clock;
+
+	/**
+	 * Constructs a factory instance that will create the {@link KafkaConsumerBackoffManager}
+	 * instances with the provided {@link KafkaConsumerTimingAdjuster}.
+	 *
+	 * @param timingAdjustmentManager the {@link KafkaConsumerTimingAdjuster} to be used.
+	 */
+	public TimingAdjustingKafkaBackOffManagerFactory(KafkaConsumerTimingAdjuster timingAdjustmentManager) {
+		this.clock = getDefaultClock();
+		setTimingAdjustmentManager(timingAdjustmentManager);
+	}
+
+	/**
+	 * Constructs a factory instance that will create the {@link KafkaConsumerBackoffManager}
+	 * instances with the provided {@link TaskExecutor} in its {@link KafkaConsumerTimingAdjuster}.
+	 *
+	 * @param timingAdjustmentManagerTaskExecutor the {@link TaskExecutor} to be used.
+	 */
+	public TimingAdjustingKafkaBackOffManagerFactory(TaskExecutor timingAdjustmentManagerTaskExecutor) {
+		this.clock = getDefaultClock();
+		setTaskExecutor(timingAdjustmentManagerTaskExecutor);
+	}
+
+	/**
+	 * Constructs a factory instance specifying whether or not timing adjustment is enabled
+	 * for this factories {@link KafkaConsumerBackoffManager}.
+	 *
+	 * @param timingAdjustmentEnabled the {@link KafkaConsumerTimingAdjuster} to be used.
+	 */
+	public TimingAdjustingKafkaBackOffManagerFactory(boolean timingAdjustmentEnabled) {
+		this.clock = getDefaultClock();
+		setTimingAdjustmentEnabled(timingAdjustmentEnabled);
+	}
+
+	/**
+	 * Constructs a factory instance using the provided {@link ListenerContainerRegistry}.
+	 *
+	 * @param listenerContainerRegistry the {@link ListenerContainerRegistry} to be used.
+	 */
+	public TimingAdjustingKafkaBackOffManagerFactory(ListenerContainerRegistry listenerContainerRegistry) {
+		super(listenerContainerRegistry);
+		this.clock = getDefaultClock();
+	}
+
+	/**
+	 * Constructs a factory instance with default dependencies.
+	 */
+	public TimingAdjustingKafkaBackOffManagerFactory() {
+		this.clock = getDefaultClock();
+	}
+
+	/**
+	 * Constructs an factory instance that will create the {@link KafkaConsumerBackoffManager}
+	 * with the provided {@link Clock}.
+	 * @param clock the clock instance to be used.
+	 */
+	public TimingAdjustingKafkaBackOffManagerFactory(Clock clock) {
+		this.clock = clock;
+	}
+
+	/**
+	 * Set this property to false if you don't want the resulting KafkaBackOffManager
+	 * to adjust the precision of the topics' consumption timing.
+	 *
+	 * @param timingAdjustmentEnabled set to false to disable timing adjustment.
+	 */
+	public final void setTimingAdjustmentEnabled(boolean timingAdjustmentEnabled) {
+		this.timingAdjustmentEnabled = timingAdjustmentEnabled;
+	}
+
+	/**
+	 * Sets the {@link WakingKafkaConsumerTimingAdjuster} that will be used
+	 * with the resulting {@link KafkaConsumerBackoffManager}.
+	 *
+	 * @param timingAdjustmentManager the adjustmentManager to be used.
+	 */
+	public final void setTimingAdjustmentManager(KafkaConsumerTimingAdjuster timingAdjustmentManager) {
+		Assert.isTrue(this.timingAdjustmentEnabled, () -> "TimingAdjustment is disabled for this factory.");
+		this.timingAdjustmentManager = timingAdjustmentManager;
+	}
+
+	/**
+	 * Sets the {@link TaskExecutor} that will be used in the {@link KafkaConsumerTimingAdjuster}.
+	 * @param taskExecutor the taskExecutor to be used.
+	 */
+	public final void setTaskExecutor(TaskExecutor taskExecutor) {
+		Assert.isTrue(this.timingAdjustmentEnabled, () -> "TimingAdjustment is disabled for this factory.");
+		this.taskExecutor = taskExecutor;
+	}
+
+	@Override
+	protected KafkaConsumerBackoffManager doCreateManager(ListenerContainerRegistry registry) {
+		KafkaConsumerBackoffManager kafkaConsumerBackoffManager = getKafkaConsumerBackoffManager(registry);
+		super.addApplicationListener(kafkaConsumerBackoffManager);
+		return kafkaConsumerBackoffManager;
+	}
+
+	protected final Clock getDefaultClock() {
+		return Clock.systemUTC();
+	}
+
+	private KafkaConsumerBackoffManager getKafkaConsumerBackoffManager(ListenerContainerRegistry registry) {
+		return this.timingAdjustmentEnabled
+			? new KafkaConsumerBackoffManager(registry, getOrCreateBackOffTimingAdjustmentManager(), this.clock)
+			: new KafkaConsumerBackoffManager(registry, this.clock);
+	}
+
+	private KafkaConsumerTimingAdjuster getOrCreateBackOffTimingAdjustmentManager() {
+		if (this.timingAdjustmentManager != null) {
+			return this.timingAdjustmentManager;
+		}
+		return new WakingKafkaConsumerTimingAdjuster(getOrCreateTimingAdjustmentThreadExecutor());
+	}
+
+	private TaskExecutor getOrCreateTimingAdjustmentThreadExecutor() {
+		if (this.taskExecutor != null) {
+			return this.taskExecutor;
+		}
+		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+		executor.initialize();
+		super.addApplicationListener((ApplicationListener<ContextClosedEvent>) event -> executor.shutdown());
+		return executor;
+	}
+}

spring-kafka/src/main/java/org/springframework/kafka/listener/WakingKafkaConsumerTimingAdjuster.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright 2018-2021 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.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.TopicPartition;
+
+import org.springframework.core.log.LogAccessor;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.retry.backoff.Sleeper;
+import org.springframework.util.Assert;
+
+
+/**
+ *
+ * Adjusts timing by creating a thread that will
+ * wakeup the consumer from polling, considering that, if consumption is paused,
+ * it will check for consumption resuming in increments of 'pollTimeout'. This works best
+ * if the consumer is handling a single partition.
+ *
+ * @author Tomaz Fernandes
+ * @since 2.7
+ * @see KafkaConsumerBackoffManager
+ */
+public class WakingKafkaConsumerTimingAdjuster implements KafkaConsumerTimingAdjuster {
+
+	private static final LogAccessor LOGGER =
+			new LogAccessor(LogFactory.getLog(WakingKafkaConsumerTimingAdjuster.class));
+
+	private static final long HUNDRED = 100L;
+
+	private static final Duration DEFAULT_TIMING_ADJUSTMENT_THRESHOLD = Duration.ofMillis(HUNDRED);
+
+	private static final int DEFAULT_POLL_TIMEOUTS_FOR_ADJUSTMENT_WINDOW = 2;
+
+	private Duration timingAdjustmentThreshold = DEFAULT_TIMING_ADJUSTMENT_THRESHOLD;
+
+	private int pollTimeoutsForAdjustmentWindow = DEFAULT_POLL_TIMEOUTS_FOR_ADJUSTMENT_WINDOW;
+
+	private final TaskExecutor timingAdjustmentTaskExecutor;
+
+	private final Sleeper sleeper;
+
+	public WakingKafkaConsumerTimingAdjuster(TaskExecutor timingAdjustmentTaskExecutor, Sleeper sleeper) {
+		Assert.notNull(timingAdjustmentTaskExecutor, "Task executor cannot be null.");
+		Assert.notNull(sleeper, "Sleeper cannot be null.");
+		this.timingAdjustmentTaskExecutor = timingAdjustmentTaskExecutor;
+		this.sleeper = sleeper;
+	}
+
+	public WakingKafkaConsumerTimingAdjuster(TaskExecutor timingAdjustmentTaskExecutor) {
+		Assert.notNull(timingAdjustmentTaskExecutor, "Task executor cannot be null.");
+		this.timingAdjustmentTaskExecutor = timingAdjustmentTaskExecutor;
+		this.sleeper = Thread::sleep;
+	}
+
+	/**
+	 *
+	 * Sets how many pollTimeouts prior to the dueTimeout the adjustment will take place.
+	 * Default is 2.
+	 *
+	 * @param pollTimeoutsForAdjustmentWindow the amount of pollTimeouts in the adjustment window.
+	 */
+	public void setPollTimeoutsForAdjustmentWindow(int pollTimeoutsForAdjustmentWindow) {
+		this.pollTimeoutsForAdjustmentWindow = pollTimeoutsForAdjustmentWindow;
+	}
+
+	/**
+	 *
+	 * Sets the threshold for the timing adjustment to take place. If the time difference between
+	 * the probable instant the message will be consumed and the instant it should is lower than
+	 * this value, no adjustment will be applied.
+	 * Default is 100ms.
+	 *
+	 * @param timingAdjustmentThreshold the threshold to be set.
+	 */
+	public void setTimingAdjustmentThreshold(Duration timingAdjustmentThreshold) {
+		this.timingAdjustmentThreshold = timingAdjustmentThreshold;
+	}
+
+	/**
+	 * Adjusts the timing with the provided parameters.
+	 *
+	 * @param consumerToAdjust the {@link Consumer} that will be adjusted
+	 * @param topicPartition the {@link TopicPartition} that will be adjusted
+	 * @param pollTimeout the pollConfiguration for the consumer's container
+	 * @param timeUntilDue the amount of time until the message is due for consumption
+	 * @return the adjusted amount in milliseconds
+	 */
+	public long adjustTiming(Consumer<?, ?> consumerToAdjust, TopicPartition topicPartition,
+							long pollTimeout, long timeUntilDue) {
+
+		boolean isInAdjustmentWindow = timeUntilDue > pollTimeout && timeUntilDue <=
+				pollTimeout * this.pollTimeoutsForAdjustmentWindow;
+
+		long adjustmentAmount = timeUntilDue % pollTimeout;
+		if (isInAdjustmentWindow && adjustmentAmount > this.timingAdjustmentThreshold.toMillis()) {
+			this.timingAdjustmentTaskExecutor.execute(() ->
+					doApplyTimingAdjustment(consumerToAdjust, topicPartition, adjustmentAmount));
+			return adjustmentAmount;
+		}
+		return 0L;
+	}
+
+	private void doApplyTimingAdjustment(Consumer<?, ?> consumerForTimingAdjustment,
+										TopicPartition topicPartition, long adjustmentAmount) {
+		try {
+			LOGGER.debug(() -> String.format("Applying timing adjustment of %s millis for TopicPartition %s",
+					adjustmentAmount, topicPartition));
+			this.sleeper.sleep(adjustmentAmount);
+			LOGGER.debug(() -> "Waking up consumer for partition topic: " + topicPartition);
+			consumerForTimingAdjustment.wakeup();
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			throw new IllegalStateException("Interrupted waking up consumer while applying timing adjustment " +
+					"for TopicPartition " + topicPartition, e);
+		}
+		catch (Exception e) { // NOSONAR
+			LOGGER.error(e, () -> "Error waking up consumer while applying timing adjustment " +
+					"for TopicPartition " + topicPartition);
+		}
+	}
+}

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java

@@ -65,12 +65,14 @@ public class ListenerContainerFactoryConfigurer {
 		CONFIGURED_FACTORIES_CACHE = new HashSet<>();
 	}
 
-	private static final int MIN_POLL_TIMEOUT_VALUE = 250;
+	private static final int MIN_POLL_TIMEOUT_VALUE = 100;
 
 	private static final int MAX_POLL_TIMEOUT_VALUE = 5000;
 
 	private static final int POLL_TIMEOUT_DIVISOR = 4;
 
+	private static final long LOWEST_BACKOFF_THRESHOLD = 1500L;
+
 	private Consumer<ConcurrentMessageListenerContainer<?, ?>> containerCustomizer = container -> {
 	};
 
@@ -85,7 +87,7 @@ public class ListenerContainerFactoryConfigurer {
 
 	ListenerContainerFactoryConfigurer(KafkaConsumerBackoffManager kafkaConsumerBackoffManager,
 									DeadLetterPublishingRecovererFactory deadLetterPublishingRecovererFactory,
-									@Qualifier(KafkaConsumerBackoffManager
+									@Qualifier(RetryTopicInternalBeanNames
 											.INTERNAL_BACKOFF_CLOCK_BEAN_NAME) Clock clock) {
 		this.kafkaConsumerBackoffManager = kafkaConsumerBackoffManager;
 		this.deadLetterPublishingRecovererFactory = deadLetterPublishingRecovererFactory;
@@ -195,7 +197,9 @@ public class ListenerContainerFactoryConfigurer {
 				.min(Comparator.naturalOrder())
 				.orElseThrow(() -> new IllegalArgumentException("No back off values found!"));
 
-		return applyLimits(lowestBackOff / POLL_TIMEOUT_DIVISOR);
+		return lowestBackOff > LOWEST_BACKOFF_THRESHOLD
+				? applyLimits(lowestBackOff / POLL_TIMEOUT_DIVISOR)
+				: MIN_POLL_TIMEOUT_VALUE;
 	}
 
 	private long applyLimits(long pollTimeoutValue) {

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicBootstrapper.java

@@ -17,14 +17,20 @@
 package org.springframework.kafka.retrytopic;
 
 import java.time.Clock;
+import java.util.function.Supplier;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.config.SingletonBeanRegistry;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.kafka.listener.KafkaBackOffManagerFactory;
 import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
+import org.springframework.kafka.listener.KafkaConsumerTimingAdjuster;
+import org.springframework.kafka.listener.TimingAdjustingKafkaBackOffManagerFactory;
 import org.springframework.retry.backoff.ThreadWaitSleeper;
 
 /**
@@ -45,26 +51,25 @@ public class RetryTopicBootstrapper {
 	private final BeanFactory beanFactory;
 
 	public RetryTopicBootstrapper(ApplicationContext applicationContext, BeanFactory beanFactory) {
-		this.beanFactory = beanFactory;
 		if (!ConfigurableApplicationContext.class.isAssignableFrom(applicationContext.getClass()) ||
-			!BeanDefinitionRegistry.class.isAssignableFrom(applicationContext.getClass())) {
+				!BeanDefinitionRegistry.class.isAssignableFrom(applicationContext.getClass())) {
 			throw new IllegalStateException(String.format("ApplicationContext must be implement %s and %s interfaces. Provided: %s",
 					ConfigurableApplicationContext.class.getSimpleName(),
 					BeanDefinitionRegistry.class.getSimpleName(),
 					applicationContext.getClass().getSimpleName()));
 		}
-		if (!SingletonBeanRegistry.class.isAssignableFrom(this.beanFactory.getClass())) {
+		if (!SingletonBeanRegistry.class.isAssignableFrom(beanFactory.getClass())) {
 			throw new IllegalStateException("BeanFactory must implement " + SingletonBeanRegistry.class +
-					" interface. Provided: " + this.beanFactory.getClass().getSimpleName());
+					" interface. Provided: " + beanFactory.getClass().getSimpleName());
 		}
+		this.beanFactory = beanFactory;
 		this.applicationContext = applicationContext;
 	}
 
 	public void bootstrapRetryTopic() {
 		registerBeans();
-		configureBackoffClock();
-		configureDestinationTopicContainer();
-		configureKafkaConsumerBackoffManager();
+		registerSingletons();
+		addApplicationListeners();
 	}
 
 	private void registerBeans() {
@@ -77,29 +82,49 @@ public class RetryTopicBootstrapper {
 		registerIfNotContains(RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_PROVIDER_NAME,
 				DeadLetterPublishingRecovererFactory.class);
 		registerIfNotContains(RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER, RetryTopicConfigurer.class);
-		registerIfNotContains(RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, KafkaConsumerBackoffManager.class);
-		registerIfNotContains(RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, DefaultDestinationTopicResolver.class);
-		registerIfNotContains(RetryTopicInternalBeanNames.DEFAULT_SLEEPER_BEAN_NAME, ThreadWaitSleeper.class);
-
+		registerIfNotContains(RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME,
+				DefaultDestinationTopicResolver.class);
+		registerIfNotContains(RetryTopicInternalBeanNames.BACKOFF_SLEEPER_BEAN_NAME, ThreadWaitSleeper.class);
+		registerIfNotContains(RetryTopicInternalBeanNames.INTERNAL_KAFKA_CONSUMER_BACKOFF_MANAGER_FACTORY,
+				TimingAdjustingKafkaBackOffManagerFactory.class);
 	}
 
-	private void configureBackoffClock() {
-		if (!this.applicationContext.containsBeanDefinition(KafkaConsumerBackoffManager.INTERNAL_BACKOFF_CLOCK_BEAN_NAME)) {
-			((SingletonBeanRegistry) this.beanFactory).registerSingleton(
-					KafkaConsumerBackoffManager.INTERNAL_BACKOFF_CLOCK_BEAN_NAME, Clock.systemUTC());
+	private void registerSingletons() {
+		registerSingletonIfNotContains(RetryTopicInternalBeanNames.INTERNAL_BACKOFF_CLOCK_BEAN_NAME, Clock::systemUTC);
+		registerSingletonIfNotContains(RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER,
+				this::createKafkaConsumerBackoffManager);
+	}
+
+	private void addApplicationListeners() {
+		((ConfigurableApplicationContext) this.applicationContext)
+				.addApplicationListener(this.applicationContext.getBean(
+						RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, DefaultDestinationTopicResolver.class));
+	}
+
+	private KafkaConsumerBackoffManager createKafkaConsumerBackoffManager() {
+		KafkaBackOffManagerFactory factory = this.applicationContext
+				.getBean(RetryTopicInternalBeanNames.INTERNAL_KAFKA_CONSUMER_BACKOFF_MANAGER_FACTORY,
+						KafkaBackOffManagerFactory.class);
+		if (ApplicationContextAware.class.isAssignableFrom(factory.getClass())) {
+			((ApplicationContextAware) factory).setApplicationContext(this.applicationContext);
 		}
+		if (TimingAdjustingKafkaBackOffManagerFactory.class.isAssignableFrom(factory.getClass())) {
+			setupTimingAdjustingBackOffFactory((TimingAdjustingKafkaBackOffManagerFactory) factory);
+		}
+		return factory.create();
 	}
 
-	private void configureKafkaConsumerBackoffManager() {
-		KafkaConsumerBackoffManager kafkaConsumerBackoffManager = this.applicationContext.getBean(
-				RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, KafkaConsumerBackoffManager.class);
-		((ConfigurableApplicationContext) this.applicationContext).addApplicationListener(kafkaConsumerBackoffManager);
-	}
-
-	private void configureDestinationTopicContainer() {
-		DefaultDestinationTopicResolver defaultDestinationTopicResolver = this.applicationContext.getBean(
-				RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, DefaultDestinationTopicResolver.class);
-		((ConfigurableApplicationContext) this.applicationContext).addApplicationListener(defaultDestinationTopicResolver);
+	private void setupTimingAdjustingBackOffFactory(TimingAdjustingKafkaBackOffManagerFactory factory) {
+		if (this.applicationContext.containsBean(RetryTopicInternalBeanNames.BACKOFF_TASK_EXECUTOR)) {
+			factory.setTaskExecutor(this.applicationContext
+					.getBean(RetryTopicInternalBeanNames.BACKOFF_TASK_EXECUTOR, TaskExecutor.class));
+		}
+		if (this.applicationContext.containsBean(
+				RetryTopicInternalBeanNames.INTERNAL_BACKOFF_TIMING_ADJUSTMENT_MANAGER)) {
+			factory.setTimingAdjustmentManager(this.applicationContext
+					.getBean(RetryTopicInternalBeanNames.INTERNAL_BACKOFF_TIMING_ADJUSTMENT_MANAGER,
+						KafkaConsumerTimingAdjuster.class));
+		}
 	}
 
 	private void registerIfNotContains(String beanName, Class<?> beanClass) {
@@ -109,4 +134,10 @@ public class RetryTopicBootstrapper {
 					new RootBeanDefinition(beanClass));
 		}
 	}
+
+	private void registerSingletonIfNotContains(String beanName, Supplier<Object> singletonSupplier) {
+		if (!this.applicationContext.containsBeanDefinition(beanName)) {
+			((SingletonBeanRegistry) this.beanFactory).registerSingleton(beanName, singletonSupplier.get());
+		}
+	}
 }

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicInternalBeanNames.java

@@ -43,9 +43,20 @@ public abstract class RetryTopicInternalBeanNames {
 
 	static final String DESTINATION_TOPIC_CONTAINER_NAME = "internalDestinationTopicContainer";
 
-	static final String DEFAULT_LISTENER_FACTORY_BEAN_NAME = "retryTopicListenerContainerFactory";
+	static final String DEFAULT_LISTENER_FACTORY_BEAN_NAME = "internalRetryTopicListenerContainerFactory";
 
-	static final String DEFAULT_SLEEPER_BEAN_NAME = "retryTopicSleeper";
+	static final String BACKOFF_SLEEPER_BEAN_NAME = "internalBackoffSleeper";
+
+	static final String BACKOFF_TASK_EXECUTOR = "internalBackOffTaskExecutor";
+
+	static final String INTERNAL_BACKOFF_TIMING_ADJUSTMENT_MANAGER = "internalKafkaConsumerTimingAdjustmentManager";
+
+	static final String INTERNAL_KAFKA_CONSUMER_BACKOFF_MANAGER_FACTORY = "internalKafkaConsumerBackOffManagerFactory";
+
+	/**
+	 * Internal Back Off Clock Bean Name.
+	 */
+	public static final String INTERNAL_BACKOFF_CLOCK_BEAN_NAME = "internalBackOffClock";
 
 	/**
 	 * Default Kafka template bean name for publishing to retry topics.

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaConsumerBackoffManagerTests.java

@@ -18,13 +18,10 @@ package org.springframework.kafka.listener;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.catchThrowableOfType;
-import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.then;
-import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 
-import java.math.BigInteger;
 import java.time.Clock;
 import java.time.Instant;
 
@@ -32,16 +29,12 @@ import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
-import org.springframework.core.task.TaskExecutor;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.event.ListenerContainerPartitionIdleEvent;
 import org.springframework.kafka.retrytopic.TestClockUtils;
-import org.springframework.retry.backoff.Sleeper;
 
 /**
  * @author Tomaz Fernandes
@@ -57,10 +50,10 @@ class KafkaConsumerBackoffManagerTests {
 	private MessageListenerContainer listenerContainer;
 
 	@Mock
-	private ListenerContainerPartitionIdleEvent partitionIdleEvent;
+	private WakingKafkaConsumerTimingAdjuster timingAdjustmentManager;
 
 	@Mock
-	private TaskExecutor taskExecutor;
+	private ListenerContainerPartitionIdleEvent partitionIdleEvent;
 
 	@Mock
 	private Consumer<?, ?> consumer;
@@ -68,12 +61,6 @@ class KafkaConsumerBackoffManagerTests {
 	@Mock
 	private ContainerProperties containerProperties;
 
-	@Mock
-	private Sleeper sleeper;
-
-	@Captor
-	private ArgumentCaptor<Runnable> correctionRunnableCaptor;
-
 	private static final String testListenerId = "testListenerId";
 
 	private static final Clock clock = TestClockUtils.CLOCK;
@@ -82,30 +69,30 @@ class KafkaConsumerBackoffManagerTests {
 
 	private static final int testPartition = 0;
 
+	private static final long pollTimeout = 500L;
+
 	private static final TopicPartition topicPartition = new TopicPartition(testTopic, testPartition);
 
 	private static final long originalTimestamp = Instant.now(clock).minusMillis(2500L).toEpochMilli();
 
-	private static final byte[] originalTimestampBytes = BigInteger.valueOf(originalTimestamp).toByteArray();
-
-
 	@Test
-	void shouldBackoffgivenDueTimestampIsLater() {
+	void shouldBackOffGivenDueTimestampIsLater() {
 
-		// setup
+		// given
 		given(this.registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
 		given(registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
-		KafkaConsumerBackoffManager backoffManager = new KafkaConsumerBackoffManager(registry, clock, taskExecutor, sleeper);
+		KafkaConsumerBackoffManager backoffManager =
+				new KafkaConsumerBackoffManager(registry, timingAdjustmentManager, clock);
 
 		long dueTimestamp = originalTimestamp + 5000;
 		KafkaConsumerBackoffManager.Context context =
 				backoffManager.createContext(dueTimestamp, testListenerId, topicPartition, consumer);
 
-		// given
+		// then
 		KafkaBackoffException backoffException = catchThrowableOfType(() -> backoffManager.maybeBackoff(context),
 				KafkaBackoffException.class);
 
-		// then
+		// when
 		assertThat(backoffException.getDueTimestamp()).isEqualTo(dueTimestamp);
 		assertThat(backoffException.getListenerId()).isEqualTo(testListenerId);
 		assertThat(backoffException.getTopicPartition()).isEqualTo(topicPartition);
@@ -116,15 +103,16 @@ class KafkaConsumerBackoffManagerTests {
 	@Test
 	void shouldNotBackoffGivenDueTimestampIsPast() {
 
-		// setup
-		KafkaConsumerBackoffManager backoffManager = new KafkaConsumerBackoffManager(registry, clock, taskExecutor, sleeper);
+		// given
+		KafkaConsumerBackoffManager backoffManager =
+				new KafkaConsumerBackoffManager(registry, timingAdjustmentManager, clock);
 		KafkaConsumerBackoffManager.Context context =
 				backoffManager.createContext(originalTimestamp - 5000, testListenerId, topicPartition, consumer);
 
-		// given
+		// then
 		backoffManager.maybeBackoff(context);
 
-		// then
+		// when
 		assertThat(backoffManager.getBackOffContext(topicPartition)).isNull();
 		then(listenerContainer).should(times(0)).pausePartition(topicPartition);
 	}
@@ -132,132 +120,92 @@ class KafkaConsumerBackoffManagerTests {
 	@Test
 	void shouldDoNothingIfIdleBeforeDueTimestamp() {
 
-		// setup
+		// given
 		given(this.partitionIdleEvent.getTopicPartition()).willReturn(topicPartition);
 		given(registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
 		given(listenerContainer.getContainerProperties()).willReturn(containerProperties);
-		given(containerProperties.getPollTimeout()).willReturn(500L);
+		given(containerProperties.getPollTimeout()).willReturn(pollTimeout);
 
-		KafkaConsumerBackoffManager backoffManager = new KafkaConsumerBackoffManager(registry, clock, taskExecutor, sleeper);
+		KafkaConsumerBackoffManager backoffManager =
+				new KafkaConsumerBackoffManager(registry, timingAdjustmentManager, clock);
 
+		long dueTimestamp = originalTimestamp + 5000;
 		KafkaConsumerBackoffManager.Context context =
-				backoffManager.createContext(originalTimestamp + 5000, testListenerId, topicPartition, consumer);
+				backoffManager.createContext(dueTimestamp, testListenerId, topicPartition, consumer);
 		backoffManager.addBackoff(context, topicPartition);
 
-		// given
+		// then
 		backoffManager.onApplicationEvent(partitionIdleEvent);
 
-		// then
+		// when
 		assertThat(backoffManager.getBackOffContext(topicPartition)).isEqualTo(context);
+		then(timingAdjustmentManager).should(times(1)).adjustTiming(
+				consumer, topicPartition, pollTimeout, getTimeUntilDue(dueTimestamp));
 		then(listenerContainer).should(times(0)).resumePartition(topicPartition);
 	}
 
+	private long getTimeUntilDue(long dueTimestamp) {
+		return dueTimestamp - Instant.now(clock).toEpochMilli();
+	}
+
 	@Test
 	void shouldResumePartitionIfIdleAfterDueTimestamp() {
 
-		// setup
+		// given
 		given(registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
 		given(listenerContainer.getContainerProperties()).willReturn(containerProperties);
 		given(containerProperties.getPollTimeout()).willReturn(500L);
 
 		given(this.registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
 		given(this.partitionIdleEvent.getTopicPartition()).willReturn(topicPartition);
-		KafkaConsumerBackoffManager backoffManager = new KafkaConsumerBackoffManager(registry, clock, taskExecutor, sleeper);
+		long dueTimestamp = originalTimestamp - 5000;
+		given(timingAdjustmentManager
+				.adjustTiming(consumer, topicPartition, pollTimeout, getTimeUntilDue(dueTimestamp)))
+				.willReturn(0L);
+		KafkaConsumerBackoffManager backoffManager =
+				new KafkaConsumerBackoffManager(registry, timingAdjustmentManager, clock);
 		KafkaConsumerBackoffManager.Context context =
-				backoffManager.createContext(originalTimestamp - 5000, testListenerId, topicPartition, consumer);
+				backoffManager.createContext(dueTimestamp, testListenerId, topicPartition, consumer);
 		backoffManager.addBackoff(context, topicPartition);
 
-		// given
+		// when
 		backoffManager.onApplicationEvent(partitionIdleEvent);
 
 		// then
+		then(timingAdjustmentManager).should(times(1)).adjustTiming(
+				consumer, topicPartition, pollTimeout, getTimeUntilDue(dueTimestamp));
 		assertThat(backoffManager.getBackOffContext(topicPartition)).isNull();
 		then(listenerContainer).should(times(1)).resumePartition(topicPartition);
 	}
 
 	@Test
-	void shouldApplyCorrectionIfInCorrectionWindow() throws InterruptedException {
+	void shouldResumePartitionIfCorrectionIsApplied() {
 
-		// setup
+		// given
 		given(registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
 		given(listenerContainer.getContainerProperties()).willReturn(containerProperties);
-		long pollTimout = 500L;
-		given(containerProperties.getPollTimeout()).willReturn(pollTimout);
+		long pollTimeout = 500L;
+		given(containerProperties.getPollTimeout()).willReturn(pollTimeout);
 
 		given(this.registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
 		given(this.partitionIdleEvent.getTopicPartition()).willReturn(topicPartition);
-		KafkaConsumerBackoffManager backoffManager = new KafkaConsumerBackoffManager(registry, clock, taskExecutor, sleeper);
-		long dueBackOffTime = 750L;
-		long dueTimestamp = Instant.now(clock).plusMillis(dueBackOffTime).toEpochMilli();
+		long dueTimestamp = originalTimestamp + 5000;
+		given(timingAdjustmentManager
+				.adjustTiming(consumer, topicPartition, pollTimeout, getTimeUntilDue(dueTimestamp)))
+				.willReturn(1000L);
+		KafkaConsumerBackoffManager backoffManager =
+				new KafkaConsumerBackoffManager(registry, timingAdjustmentManager, clock);
 		KafkaConsumerBackoffManager.Context context =
-				backoffManager.createContext(dueTimestamp,
-						testListenerId, topicPartition, consumer);
+				backoffManager.createContext(dueTimestamp, testListenerId, topicPartition, consumer);
 		backoffManager.addBackoff(context, topicPartition);
 
-		// given
+		// when
 		backoffManager.onApplicationEvent(partitionIdleEvent);
 
 		// then
-		then(this.taskExecutor).should(times(1)).execute(correctionRunnableCaptor.capture());
-		Runnable correctionRunnable = correctionRunnableCaptor.getValue();
-		correctionRunnable.run();
-
-		then(sleeper).should(times(1)).sleep(dueBackOffTime - pollTimout);
-		then(consumer).should(times(1)).wakeup();
-
-		assertThat(backoffManager.getBackOffContext(topicPartition)).isNull();
+		then(timingAdjustmentManager).should(times(1)).adjustTiming(
+				consumer, topicPartition, pollTimeout, getTimeUntilDue(dueTimestamp));
 		then(listenerContainer).should(times(1)).resumePartition(topicPartition);
-	}
-
-	@Test
-	void shouldNotApplyCorrectionIfTooLateForCorrectionWindow() throws InterruptedException {
-
-		// setup
-		given(registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
-		given(listenerContainer.getContainerProperties()).willReturn(containerProperties);
-		long pollTimout = 500L;
-		given(containerProperties.getPollTimeout()).willReturn(pollTimout);
-
-		given(this.registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
-		given(this.partitionIdleEvent.getTopicPartition()).willReturn(topicPartition);
-		KafkaConsumerBackoffManager backoffManager = new KafkaConsumerBackoffManager(registry, clock, taskExecutor, sleeper);
-		long dueBackOffTime = 250L;
-		long dueTimestamp = Instant.now(clock).plusMillis(dueBackOffTime).toEpochMilli();
-		KafkaConsumerBackoffManager.Context context =
-				backoffManager.createContext(dueTimestamp,
-						testListenerId, topicPartition, consumer);
-		backoffManager.addBackoff(context, topicPartition);
-
-		// given
-		backoffManager.onApplicationEvent(partitionIdleEvent);
-
-		// then
-		then(this.taskExecutor).should(never()).execute(any(Runnable.class));
-	}
-
-	@Test
-	void shouldNotApplyCorrectionIfTooSoonForCorrectionWindow() throws InterruptedException {
-
-		// setup
-		given(registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
-		given(listenerContainer.getContainerProperties()).willReturn(containerProperties);
-		long pollTimout = 500L;
-		given(containerProperties.getPollTimeout()).willReturn(pollTimout);
-
-		given(this.registry.getListenerContainer(testListenerId)).willReturn(listenerContainer);
-		given(this.partitionIdleEvent.getTopicPartition()).willReturn(topicPartition);
-		KafkaConsumerBackoffManager backoffManager = new KafkaConsumerBackoffManager(registry, clock, taskExecutor, sleeper);
-		long dueBackOffTime = 1250L;
-		long dueTimestamp = Instant.now(clock).plusMillis(dueBackOffTime).toEpochMilli();
-		KafkaConsumerBackoffManager.Context context =
-				backoffManager.createContext(dueTimestamp,
-						testListenerId, topicPartition, consumer);
-		backoffManager.addBackoff(context, topicPartition);
-
-		// given
-		backoffManager.onApplicationEvent(partitionIdleEvent);
-
-		// then
-		then(this.taskExecutor).should(never()).execute(any(Runnable.class));
+		assertThat(backoffManager.getBackOffContext(topicPartition)).isNull();
 	}
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/WakingKafkaConsumerTimingAdjusterTests.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2019-2021 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 static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.then;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.retry.backoff.Sleeper;
+
+/**
+ * @author Tomaz Fernandes
+ * @since 2.7
+ */
+class WakingKafkaConsumerTimingAdjusterTests {
+
+	@Test
+	void testAppliesCorrectionIfInCorrectionWindow() throws InterruptedException {
+		Sleeper sleeper = mock(Sleeper.class);
+		TaskExecutor taskExecutor = mock(TaskExecutor.class);
+		Consumer<?, ?> consumer = mock(Consumer.class);
+		TopicPartition topicPartition = new TopicPartition("test-topic", 0);
+		long pollTimout = 500L;
+		long dueBackOffTime = 750L;
+		ArgumentCaptor<Runnable> correctionRunnableCaptor = ArgumentCaptor.forClass(Runnable.class);
+		WakingKafkaConsumerTimingAdjuster timingAdjuster = new WakingKafkaConsumerTimingAdjuster(taskExecutor, sleeper);
+		timingAdjuster.adjustTiming(consumer, topicPartition, pollTimout, dueBackOffTime);
+		then(taskExecutor).should(times(1)).execute(correctionRunnableCaptor.capture());
+		Runnable correctionRunnable = correctionRunnableCaptor.getValue();
+		correctionRunnable.run();
+		then(sleeper).should(times(1)).sleep(dueBackOffTime - pollTimout);
+		then(consumer).should(times(1)).wakeup();
+	}
+
+	@Test
+	void testDoesNotApplyCorrectionIfTooLateForCorrectionWindow() {
+		Sleeper sleeper = mock(Sleeper.class);
+		TaskExecutor taskExecutor = mock(TaskExecutor.class);
+		Consumer<?, ?> consumer = mock(Consumer.class);
+		TopicPartition topicPartition = new TopicPartition("test-topic", 0);
+		long pollTimout = 500L;
+		long dueBackOffTime = 250L;
+		WakingKafkaConsumerTimingAdjuster timingAdjuster = new WakingKafkaConsumerTimingAdjuster(taskExecutor, sleeper);
+		timingAdjuster.adjustTiming(consumer, topicPartition, pollTimout, dueBackOffTime);
+		then(taskExecutor).should(never()).execute(any(Runnable.class));
+	}
+
+	@Test
+	void testDoesNotApplyCorrectionIfTooSoonForCorrectionWindow() {
+		Sleeper sleeper = mock(Sleeper.class);
+		TaskExecutor taskExecutor = mock(TaskExecutor.class);
+		Consumer<?, ?> consumer = mock(Consumer.class);
+		TopicPartition topicPartition = new TopicPartition("test-topic", 0);
+		long pollTimout = 500L;
+		long dueBackOffTime = 1250L;
+		WakingKafkaConsumerTimingAdjuster timingAdjuster = new WakingKafkaConsumerTimingAdjuster(taskExecutor, sleeper);
+		timingAdjuster.adjustTiming(consumer, topicPartition, pollTimout, dueBackOffTime);
+		then(taskExecutor).should(never()).execute(any(Runnable.class));
+	}
+}

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java

@@ -132,7 +132,7 @@ class ListenerContainerFactoryConfigurerTests {
 	@Mock
 	private RetryTopicConfiguration configuration;
 
-	private final long backOffValue = 1000L;
+	private final long backOffValue = 2000L;
 
 	private ListenerContainerFactoryConfigurer.Configuration lcfcConfiguration =
 			new ListenerContainerFactoryConfigurer.Configuration(Collections.singletonList(backOffValue));
@@ -254,9 +254,9 @@ class ListenerContainerFactoryConfigurerTests {
 		containerCustomizer.configure(container);
 
 		then(containerProperties).should(times(1))
-				.setIdlePartitionEventInterval(250L);
+				.setIdlePartitionEventInterval(100L);
 		then(containerProperties).should(times(1))
-				.setPollTimeout(250L);
+				.setPollTimeout(100L);
 	}
 
 	@Test

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicBootstrapperTests.java

@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.then;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 
 import java.time.Clock;
@@ -31,13 +32,13 @@ import org.mockito.junit.jupiter.MockitoExtension;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.config.SingletonBeanRegistry;
-import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.context.ApplicationContext;
-import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.kafka.listener.KafkaBackOffManagerFactory;
 import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
+import org.springframework.kafka.listener.TimingAdjustingKafkaBackOffManagerFactory;
 import org.springframework.retry.backoff.ThreadWaitSleeper;
 
 /**
@@ -63,7 +64,10 @@ class RetryTopicBootstrapperTests {
 	private DefaultDestinationTopicResolver defaultDestinationTopicResolver;
 
 	@Mock
-	private KafkaConsumerBackoffManager kafkaConsumerBackoffManager;
+	private TimingAdjustingKafkaBackOffManagerFactory kafkaBackOffManagerFactory;
+
+	@Mock
+	private KafkaConsumerBackoffManager kafkaConsumerBackOffManager;
 
 	@Test
 	void shouldThrowIfACDoesntImplementInterfaces() {
@@ -86,24 +90,36 @@ class RetryTopicBootstrapperTests {
 				RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, DefaultDestinationTopicResolver.class))
 				.willReturn(defaultDestinationTopicResolver);
 		given(this.applicationContext.getBean(
-				RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, KafkaConsumerBackoffManager.class))
-				.willReturn(kafkaConsumerBackoffManager);
-
-		BeanDefinitionRegistry registry = (BeanDefinitionRegistry) this.applicationContext;
+				RetryTopicInternalBeanNames.INTERNAL_KAFKA_CONSUMER_BACKOFF_MANAGER_FACTORY,
+				KafkaBackOffManagerFactory.class))
+				.willReturn(kafkaBackOffManagerFactory);
 
 		// when
 		RetryTopicBootstrapper bootstrapper = new RetryTopicBootstrapper(applicationContext, beanFactory);
 		bootstrapper.bootstrapRetryTopic();
 
 		// then
-		then(registry).should(times(1)).registerBeanDefinition(RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_RESOLVER_NAME, new RootBeanDefinition(ListenerContainerFactoryResolver.class));
-		then(registry).should(times(1)).registerBeanDefinition(RetryTopicInternalBeanNames.DESTINATION_TOPIC_PROCESSOR_NAME, new RootBeanDefinition(DefaultDestinationTopicProcessor.class));
-		then(registry).should(times(1)).registerBeanDefinition(RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME, new RootBeanDefinition(ListenerContainerFactoryConfigurer.class));
-		then(registry).should(times(1)).registerBeanDefinition(RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_PROVIDER_NAME, new RootBeanDefinition(DeadLetterPublishingRecovererFactory.class));
-		then(registry).should(times(1)).registerBeanDefinition(RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER, new RootBeanDefinition(RetryTopicConfigurer.class));
-		then(registry).should(times(1)).registerBeanDefinition(RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, new RootBeanDefinition(KafkaConsumerBackoffManager.class));
-		then(registry).should(times(1)).registerBeanDefinition(RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, new RootBeanDefinition(DefaultDestinationTopicResolver.class));
-		then(registry).should(times(1)).registerBeanDefinition(RetryTopicInternalBeanNames.DEFAULT_SLEEPER_BEAN_NAME, new RootBeanDefinition(ThreadWaitSleeper.class));
+		then(this.applicationContext).should(times(1))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_RESOLVER_NAME,
+						new RootBeanDefinition(ListenerContainerFactoryResolver.class));
+		then(this.applicationContext).should(times(1))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.DESTINATION_TOPIC_PROCESSOR_NAME,
+						new RootBeanDefinition(DefaultDestinationTopicProcessor.class));
+		then(this.applicationContext).should(times(1))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME,
+						new RootBeanDefinition(ListenerContainerFactoryConfigurer.class));
+		then(this.applicationContext).should(times(1))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_PROVIDER_NAME,
+						new RootBeanDefinition(DeadLetterPublishingRecovererFactory.class));
+		then(this.applicationContext).should(times(1))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER,
+						new RootBeanDefinition(RetryTopicConfigurer.class));
+		then(this.applicationContext).should(times(1))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME,
+						new RootBeanDefinition(DefaultDestinationTopicResolver.class));
+		then(this.applicationContext).should(times(1))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.BACKOFF_SLEEPER_BEAN_NAME,
+						new RootBeanDefinition(ThreadWaitSleeper.class));
 	}
 
 	@Test
@@ -114,41 +130,45 @@ class RetryTopicBootstrapperTests {
 		given(this.applicationContext.getBean(
 				RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, DefaultDestinationTopicResolver.class))
 				.willReturn(defaultDestinationTopicResolver);
-		given(this.applicationContext.getBean(
-				RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, KafkaConsumerBackoffManager.class))
-				.willReturn(kafkaConsumerBackoffManager);
-
-		BeanDefinitionRegistry registry = (BeanDefinitionRegistry) this.applicationContext;
 
 		// when
 		RetryTopicBootstrapper bootstrapper = new RetryTopicBootstrapper(applicationContext, beanFactory);
 		bootstrapper.bootstrapRetryTopic();
 
 		// then
-		then(registry).should(times(0)).registerBeanDefinition(RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_RESOLVER_NAME, new RootBeanDefinition(ListenerContainerFactoryResolver.class));
-		then(registry).should(times(0)).registerBeanDefinition(RetryTopicInternalBeanNames.DESTINATION_TOPIC_PROCESSOR_NAME, new RootBeanDefinition(DefaultDestinationTopicProcessor.class));
-		then(registry).should(times(0)).registerBeanDefinition(RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME, new RootBeanDefinition(ListenerContainerFactoryConfigurer.class));
-		then(registry).should(times(0)).registerBeanDefinition(RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_PROVIDER_NAME, new RootBeanDefinition(DeadLetterPublishingRecovererFactory.class));
-		then(registry).should(times(0)).registerBeanDefinition(RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER, new RootBeanDefinition(RetryTopicConfigurer.class));
-		then(registry).should(times(0)).registerBeanDefinition(RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, new RootBeanDefinition(KafkaConsumerBackoffManager.class));
-		then(registry).should(times(0)).registerBeanDefinition(RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, new RootBeanDefinition(DefaultDestinationTopicResolver.class));
-		then(registry).should(times(0)).registerBeanDefinition(RetryTopicInternalBeanNames.DEFAULT_SLEEPER_BEAN_NAME, new RootBeanDefinition(ThreadWaitSleeper.class));
+		then(this.applicationContext).should(times(0))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.DESTINATION_TOPIC_PROCESSOR_NAME,
+						new RootBeanDefinition(DefaultDestinationTopicProcessor.class));
+		then(this.applicationContext).should(times(0))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME,
+						new RootBeanDefinition(ListenerContainerFactoryConfigurer.class));
+		then(this.applicationContext).should(times(0))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_RESOLVER_NAME,
+						new RootBeanDefinition(ListenerContainerFactoryResolver.class));
+		then(this.applicationContext).should(times(0))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_PROVIDER_NAME,
+						new RootBeanDefinition(DeadLetterPublishingRecovererFactory.class));
+		then(this.applicationContext).should(times(0))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER,
+						new RootBeanDefinition(RetryTopicConfigurer.class));
+		then(this.applicationContext).should(times(0))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME,
+						new RootBeanDefinition(DefaultDestinationTopicResolver.class));
+		then(this.applicationContext).should(times(0))
+				.registerBeanDefinition(RetryTopicInternalBeanNames.BACKOFF_SLEEPER_BEAN_NAME,
+						new RootBeanDefinition(ThreadWaitSleeper.class));
 	}
 
 	@Test
-	void shouldConfigureClock() {
+	void shouldRegisterSingletonsIfNotExists() {
 
 		// given
 		given(applicationContext.containsBeanDefinition(any(String.class)))
 				.willReturn(false);
-		given(this.applicationContext.getBean(
-				RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, DefaultDestinationTopicResolver.class))
-				.willReturn(defaultDestinationTopicResolver);
-		given(this.applicationContext.getBean(
-				RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, KafkaConsumerBackoffManager.class))
-				.willReturn(kafkaConsumerBackoffManager);
-
-		BeanDefinitionRegistry registry = (BeanDefinitionRegistry) this.applicationContext;
+		given(this.applicationContext
+				.getBean(RetryTopicInternalBeanNames.INTERNAL_KAFKA_CONSUMER_BACKOFF_MANAGER_FACTORY,
+					KafkaBackOffManagerFactory.class)).willReturn(kafkaBackOffManagerFactory);
+		given(kafkaBackOffManagerFactory.create()).willReturn(kafkaConsumerBackOffManager);
 
 		// when
 		RetryTopicBootstrapper bootstrapper = new RetryTopicBootstrapper(applicationContext, beanFactory);
@@ -156,7 +176,31 @@ class RetryTopicBootstrapperTests {
 
 		// then
 		then((SingletonBeanRegistry) this.beanFactory).should(times(1)).registerSingleton(
-				KafkaConsumerBackoffManager.INTERNAL_BACKOFF_CLOCK_BEAN_NAME, Clock.systemUTC());
+				RetryTopicInternalBeanNames.INTERNAL_BACKOFF_CLOCK_BEAN_NAME, Clock.systemUTC());
+		then((SingletonBeanRegistry) this.beanFactory).should(times(1)).registerSingleton(
+				RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, kafkaConsumerBackOffManager);
+	}
+
+	@Test
+	void shouldNotRegisterSingletonsIfExists() {
+
+		// given
+		given(applicationContext.containsBeanDefinition(any(String.class)))
+				.willReturn(false);
+		given(applicationContext.containsBeanDefinition(RetryTopicInternalBeanNames.INTERNAL_BACKOFF_CLOCK_BEAN_NAME))
+				.willReturn(true);
+		given(applicationContext.containsBeanDefinition(RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER))
+				.willReturn(true);
+
+		// when
+		RetryTopicBootstrapper bootstrapper = new RetryTopicBootstrapper(applicationContext, beanFactory);
+		bootstrapper.bootstrapRetryTopic();
+
+		// then
+		then((SingletonBeanRegistry) this.beanFactory).should(never()).registerSingleton(
+				RetryTopicInternalBeanNames.INTERNAL_BACKOFF_CLOCK_BEAN_NAME, Clock.systemUTC());
+		then((SingletonBeanRegistry) this.beanFactory).should(never()).registerSingleton(
+				RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, kafkaConsumerBackOffManager);
 	}
 
 	@Test
@@ -169,18 +213,16 @@ class RetryTopicBootstrapperTests {
 				RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, DefaultDestinationTopicResolver.class))
 				.willReturn(defaultDestinationTopicResolver);
 		given(this.applicationContext.getBean(
-				RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, KafkaConsumerBackoffManager.class))
-				.willReturn(kafkaConsumerBackoffManager);
-
-		BeanDefinitionRegistry registry = (BeanDefinitionRegistry) this.applicationContext;
+				RetryTopicInternalBeanNames.INTERNAL_KAFKA_CONSUMER_BACKOFF_MANAGER_FACTORY,
+				KafkaBackOffManagerFactory.class))
+				.willReturn(kafkaBackOffManagerFactory);
 
 		// when
 		RetryTopicBootstrapper bootstrapper = new RetryTopicBootstrapper(applicationContext, beanFactory);
 		bootstrapper.bootstrapRetryTopic();
-		ConfigurableApplicationContext configurableApplicationContext = (ConfigurableApplicationContext) this.applicationContext;
 
 		// then
-		then(configurableApplicationContext).should(times(1)).addApplicationListener(kafkaConsumerBackoffManager);
-		then(configurableApplicationContext).should(times(1)).addApplicationListener(defaultDestinationTopicResolver);
+		then(this.applicationContext).should(times(1))
+				.addApplicationListener(defaultDestinationTopicResolver);
 	}
 }

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicInternalBeanNamesTests.java

@@ -40,7 +40,7 @@ class RetryTopicInternalBeanNamesTests {
 
 	static final String DESTINATION_TOPIC_CONTAINER_NAME = "internalDestinationTopicContainer";
 
-	static final String DEFAULT_LISTENER_FACTORY_BEAN_NAME = "retryTopicListenerContainerFactory";
+	static final String DEFAULT_LISTENER_FACTORY_BEAN_NAME = "internalRetryTopicListenerContainerFactory";
 
 	static final String DEFAULT_KAFKA_TEMPLATE_BEAN_NAME = "retryTopicDefaultKafkaTemplate";
 

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicIntegrationTests.java

@@ -53,7 +53,6 @@ import org.springframework.kafka.core.KafkaAdmin;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.ContainerProperties;
-import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.context.EmbeddedKafka;
@@ -371,7 +370,7 @@ public class RetryableTopicIntegrationTests {
 	@Configuration
 	public static class RuntimeConfig {
 
-		@Bean(name = KafkaConsumerBackoffManager.INTERNAL_BACKOFF_CLOCK_BEAN_NAME)
+		@Bean(name = RetryTopicInternalBeanNames.INTERNAL_BACKOFF_CLOCK_BEAN_NAME)
 		public Clock clock() {
 			return Clock.systemUTC();
 		}