From 290d11a319145a741303f6affcff23ff66ee57d2 Mon Sep 17 00:00:00 2001 From: Tomaz Fernandes Date: Sun, 4 Apr 2021 21:43:30 -0300 Subject: [PATCH] GH-1744: Improve KafkaConsumerBackoffManager dependency management Resolves #1744 Introduces the KafkaConsumerTimingAdjuster interface Introduces the KafkaBackOffManagerFactory interface Polish code as per code review Polishing. --- .../AbstractKafkaBackOffManagerFactory.java | 98 ++++++++++ .../listener/KafkaBackOffManagerFactory.java | 31 ++++ .../listener/KafkaConsumerBackoffManager.java | 168 ++++++++++-------- .../listener/KafkaConsumerTimingAdjuster.java | 46 +++++ ...ngAdjustingKafkaBackOffManagerFactory.java | 167 +++++++++++++++++ .../WakingKafkaConsumerTimingAdjuster.java | 141 +++++++++++++++ .../ListenerContainerFactoryConfigurer.java | 10 +- .../retrytopic/RetryTopicBootstrapper.java | 81 ++++++--- .../RetryTopicInternalBeanNames.java | 15 +- .../KafkaConsumerBackoffManagerTests.java | 162 ++++++----------- ...akingKafkaConsumerTimingAdjusterTests.java | 82 +++++++++ ...stenerContainerFactoryConfigurerTests.java | 6 +- .../RetryTopicBootstrapperTests.java | 132 +++++++++----- .../RetryTopicInternalBeanNamesTests.java | 2 +- .../RetryableTopicIntegrationTests.java | 3 +- 15 files changed, 882 insertions(+), 262 deletions(-) create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractKafkaBackOffManagerFactory.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaBackOffManagerFactory.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerTimingAdjuster.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/TimingAdjustingKafkaBackOffManagerFactory.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/listener/WakingKafkaConsumerTimingAdjuster.java create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/listener/WakingKafkaConsumerTimingAdjusterTests.java diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractKafkaBackOffManagerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractKafkaBackOffManagerFactory.java new file mode 100644 index 00000000..65debf3d --- /dev/null +++ b/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 getBean(String beanName, Class 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; + } +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaBackOffManagerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaBackOffManagerFactory.java new file mode 100644 index 00000000..1f641a67 --- /dev/null +++ b/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(); + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerBackoffManager.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerBackoffManager.java index 11448cad..313b39aa 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerBackoffManager.java +++ b/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 { 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 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 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 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 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 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; } } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerTimingAdjuster.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerTimingAdjuster.java new file mode 100644 index 00000000..4dd5e0ab --- /dev/null +++ b/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); +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/TimingAdjustingKafkaBackOffManagerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/TimingAdjustingKafkaBackOffManagerFactory.java new file mode 100644 index 00000000..2d259e9f --- /dev/null +++ b/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) event -> executor.shutdown()); + return executor; + } +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/WakingKafkaConsumerTimingAdjuster.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/WakingKafkaConsumerTimingAdjuster.java new file mode 100644 index 00000000..0ccbd8f2 --- /dev/null +++ b/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); + } + } +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java index 6b959d3a..2447d1f0 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java +++ b/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> 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) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicBootstrapper.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicBootstrapper.java index 1ffb953b..8c580a07 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicBootstrapper.java +++ b/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 singletonSupplier) { + if (!this.applicationContext.containsBeanDefinition(beanName)) { + ((SingletonBeanRegistry) this.beanFactory).registerSingleton(beanName, singletonSupplier.get()); + } + } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicInternalBeanNames.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicInternalBeanNames.java index 7d46563a..27473f09 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicInternalBeanNames.java +++ b/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. diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaConsumerBackoffManagerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaConsumerBackoffManagerTests.java index 65535081..2044cc89 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaConsumerBackoffManagerTests.java +++ b/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 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(); } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/WakingKafkaConsumerTimingAdjusterTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/WakingKafkaConsumerTimingAdjusterTests.java new file mode 100644 index 00000000..837d5921 --- /dev/null +++ b/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 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)); + } +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java index c6a83626..6e5fd445 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java +++ b/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 diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicBootstrapperTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicBootstrapperTests.java index 594df337..e2ed67ce 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicBootstrapperTests.java +++ b/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); } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicInternalBeanNamesTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicInternalBeanNamesTests.java index 16965305..4775c37b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicInternalBeanNamesTests.java +++ b/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"; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicIntegrationTests.java index b32d3442..bc5a65d5 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicIntegrationTests.java +++ b/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(); }