GH-1762: Extract KafkaConsumerBackoffManager

Resolves #1762

Remove unnecessary NOSONAR comments.

Author: tomazfernandes

spring-kafka/src/main/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * 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.

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

@@ -16,255 +16,80 @@
 
 package org.springframework.kafka.listener;
 
-import java.time.Clock;
-import java.time.Instant;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
 
-import org.springframework.context.ApplicationListener;
-import org.springframework.core.log.LogAccessor;
-import org.springframework.kafka.event.ListenerContainerPartitionIdleEvent;
 import org.springframework.lang.Nullable;
 
 /**
- *
- * A manager that backs off consumption for a given topic if the timestamp provided is not
- * due. Use with {@link SeekToCurrentErrorHandler} to guarantee that the message is read
- * again after partition consumption is resumed (or seek it manually by other means).
- * It's also necessary to set a {@link ContainerProperties#setIdlePartitionEventInterval(Long)}
- * 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.
+ * Interface for backing off a {@link MessageListenerContainer}
+ * until a given dueTimestamp, if such timestamp is in the future.
  *
  * @author Tomaz Fernandes
  * @author Gary Russell
  * @since 2.7
- * @see SeekToCurrentErrorHandler
  */
-public class KafkaConsumerBackoffManager implements ApplicationListener<ListenerContainerPartitionIdleEvent> {
-
-	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(KafkaConsumerBackoffManager.class));
-
-	private final ListenerContainerRegistry listenerContainerRegistry;
+public interface KafkaConsumerBackoffManager {
 
-	private final Map<TopicPartition, Context> backOffContexts;
-
-	private final Clock clock;
-
-	private final KafkaConsumerTimingAdjuster kafkaConsumerTimingAdjuster;
-
-	/**
-	 * 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) {
+	void backOffIfNecessary(Context context);
 
-		this.listenerContainerRegistry = listenerContainerRegistry;
-		this.kafkaConsumerTimingAdjuster = kafkaConsumerTimingAdjuster;
-		this.clock = Clock.systemUTC();
-		this.backOffContexts = new HashMap<>();
-	}
-
-	/**
-	 * 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.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<>();
-	}
-
-	/**
-	 * Backs off if the current time is before the dueTimestamp provided
-	 * in the {@link Context} object.
-	 * @param context the back off context for this execution.
-	 */
-	public void maybeBackoff(Context context) {
-		long backoffTime = context.dueTimestamp - getCurrentMillisFromClock();
-		if (backoffTime > 0) {
-			pauseConsumptionAndThrow(context, backoffTime);
-		}
-	}
-
-	private void pauseConsumptionAndThrow(Context context, Long backOffTime) throws KafkaBackoffException {
-		TopicPartition topicPartition = context.topicPartition;
-		getListenerContainerFromContext(context).pausePartition(topicPartition);
-		addBackoff(context, topicPartition);
-		throw new KafkaBackoffException(String.format("Partition %s from topic %s is not ready for consumption, " +
-				"backing off for approx. %s millis.", context.topicPartition.partition(),
-				context.topicPartition.topic(), backOffTime),
-				topicPartition, context.listenerId, context.dueTimestamp);
-	}
-
-	@Override
-	public void onApplicationEvent(ListenerContainerPartitionIdleEvent partitionIdleEvent) {
-		LOGGER.debug(() -> String.format("partitionIdleEvent received at %s. Partition: %s",
-				getCurrentMillisFromClock(), partitionIdleEvent.getTopicPartition()));
-
-		Context backOffContext = getBackOffContext(partitionIdleEvent.getTopicPartition());
-		maybeResumeConsumption(backOffContext);
-	}
-
-	private long getCurrentMillisFromClock() {
-		return Instant.now(this.clock).toEpochMilli();
-	}
-
-	private void maybeResumeConsumption(@Nullable Context context) {
-		if (context == null) {
-			return;
-		}
-		long now = getCurrentMillisFromClock();
-		long timeUntilDue = context.dueTimestamp - now;
-		long pollTimeout = getListenerContainerFromContext(context)
-				.getContainerProperties()
-				.getPollTimeout();
-		boolean isDue = timeUntilDue <= pollTimeout;
-
-		long adjustedAmount = applyTimingAdjustment(context, timeUntilDue, pollTimeout);
-
-		if (adjustedAmount != 0L || isDue) {
-			resumePartition(context);
-		}
-		else {
-			LOGGER.debug(() -> String.format("TopicPartition %s not due. DueTimestamp: %s Now: %s ",
-					context.topicPartition, context.dueTimestamp, now));
-		}
-	}
-
-	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());
-		container.resumePartition(context.topicPartition);
-		removeBackoff(context.topicPartition);
-	}
-
-	private MessageListenerContainer getListenerContainerFromContext(Context context) {
-		return this.listenerContainerRegistry.getListenerContainer(context.listenerId);
-	}
-
-	protected void addBackoff(Context context, TopicPartition topicPartition) {
-		synchronized (this.backOffContexts) {
-			this.backOffContexts.put(topicPartition, context);
-		}
-	}
-
-	protected @Nullable Context getBackOffContext(TopicPartition topicPartition) {
-		synchronized (this.backOffContexts) {
-			return this.backOffContexts.get(topicPartition);
-		}
-	}
-
-	protected void removeBackoff(TopicPartition topicPartition) {
-		synchronized (this.backOffContexts) {
-			this.backOffContexts.remove(topicPartition);
-		}
-	}
-
-	public Context createContext(long dueTimestamp, String listenerId, TopicPartition topicPartition,
-								@Nullable Consumer<?, ?> consumerForTimingAdjustment) {
-		return new Context(dueTimestamp, topicPartition, listenerId, consumerForTimingAdjustment);
+	default Context createContext(long dueTimestamp, String listenerId, TopicPartition topicPartition,
+								@Nullable Consumer<?, ?> messageConsumer) {
+		return new Context(dueTimestamp, topicPartition, listenerId, messageConsumer);
 	}
 
 	/**
 	 * Provides the state that will be used for backing off.
 	 * @since 2.7
 	 */
-	public static class Context {
+	class Context {
 
 		/**
 		 * The time after which the message should be processed,
 		 * in milliseconds since epoch.
 		 */
-		private final long dueTimestamp; // NOSONAR
+		private final long dueTimestamp;
 
 		/**
 		 * The id for the listener that should be paused.
 		 */
-		private final String listenerId; // NOSONAR
+		private final String listenerId;
 
 		/**
 		 * The topic that contains the partition to be paused.
 		 */
-		private final TopicPartition topicPartition; // NOSONAR
+		private final TopicPartition topicPartition;
 
 		/**
 		 * The consumer of the message, if present.
 		 */
-		private final Consumer<?, ?> consumerForTimingAdjustment; // NOSONAR
+		private final Consumer<?, ?> consumerForTimingAdjustment;
 
 		Context(long dueTimestamp, TopicPartition topicPartition, String listenerId,
-						@Nullable Consumer<?, ?> consumerForTimingAdjustment) {
+				@Nullable Consumer<?, ?> consumerForTimingAdjustment) {
+
 			this.dueTimestamp = dueTimestamp;
 			this.listenerId = listenerId;
 			this.topicPartition = topicPartition;
 			this.consumerForTimingAdjustment = consumerForTimingAdjustment;
 		}
+
+		public long getDueTimestamp() {
+			return this.dueTimestamp;
+		}
+
+		public String getListenerId() {
+			return this.listenerId;
+		}
+
+		public TopicPartition getTopicPartition() {
+			return this.topicPartition;
+		}
+
+		public @Nullable Consumer<?, ?> getConsumerForTimingAdjustment() {
+			return this.consumerForTimingAdjustment;
+		}
+
 	}
+
 }

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

@@ -32,7 +32,7 @@
  * @author Tomaz Fernandes
  * @since 2.7
  */
-public class TimingAdjustingKafkaBackOffManagerFactory extends AbstractKafkaBackOffManagerFactory {
+public class PartitionPausingBackOffManagerFactory extends AbstractKafkaBackOffManagerFactory {
 
 	private boolean timingAdjustmentEnabled = true;
 
@@ -48,7 +48,7 @@ public class TimingAdjustingKafkaBackOffManagerFactory extends AbstractKafkaBack
 	 *
 	 * @param timingAdjustmentManager the {@link KafkaConsumerTimingAdjuster} to be used.
 	 */
-	public TimingAdjustingKafkaBackOffManagerFactory(KafkaConsumerTimingAdjuster timingAdjustmentManager) {
+	public PartitionPausingBackOffManagerFactory(KafkaConsumerTimingAdjuster timingAdjustmentManager) {
 		this.clock = getDefaultClock();
 		setTimingAdjustmentManager(timingAdjustmentManager);
 	}
@@ -59,7 +59,7 @@ public TimingAdjustingKafkaBackOffManagerFactory(KafkaConsumerTimingAdjuster tim
 	 *
 	 * @param timingAdjustmentManagerTaskExecutor the {@link TaskExecutor} to be used.
 	 */
-	public TimingAdjustingKafkaBackOffManagerFactory(TaskExecutor timingAdjustmentManagerTaskExecutor) {
+	public PartitionPausingBackOffManagerFactory(TaskExecutor timingAdjustmentManagerTaskExecutor) {
 		this.clock = getDefaultClock();
 		setTaskExecutor(timingAdjustmentManagerTaskExecutor);
 	}
@@ -70,7 +70,7 @@ public TimingAdjustingKafkaBackOffManagerFactory(TaskExecutor timingAdjustmentMa
 	 *
 	 * @param timingAdjustmentEnabled the {@link KafkaConsumerTimingAdjuster} to be used.
 	 */
-	public TimingAdjustingKafkaBackOffManagerFactory(boolean timingAdjustmentEnabled) {
+	public PartitionPausingBackOffManagerFactory(boolean timingAdjustmentEnabled) {
 		this.clock = getDefaultClock();
 		setTimingAdjustmentEnabled(timingAdjustmentEnabled);
 	}
@@ -80,15 +80,15 @@ public TimingAdjustingKafkaBackOffManagerFactory(boolean timingAdjustmentEnabled
 	 *
 	 * @param listenerContainerRegistry the {@link ListenerContainerRegistry} to be used.
 	 */
-	public TimingAdjustingKafkaBackOffManagerFactory(ListenerContainerRegistry listenerContainerRegistry) {
+	public PartitionPausingBackOffManagerFactory(ListenerContainerRegistry listenerContainerRegistry) {
 		super(listenerContainerRegistry);
 		this.clock = getDefaultClock();
 	}
 
 	/**
 	 * Constructs a factory instance with default dependencies.
 	 */
-	public TimingAdjustingKafkaBackOffManagerFactory() {
+	public PartitionPausingBackOffManagerFactory() {
 		this.clock = getDefaultClock();
 	}
 
@@ -97,7 +97,7 @@ public TimingAdjustingKafkaBackOffManagerFactory() {
 	 * with the provided {@link Clock}.
 	 * @param clock the clock instance to be used.
 	 */
-	public TimingAdjustingKafkaBackOffManagerFactory(Clock clock) {
+	public PartitionPausingBackOffManagerFactory(Clock clock) {
 		this.clock = clock;
 	}
 
@@ -133,7 +133,7 @@ public final void setTaskExecutor(TaskExecutor taskExecutor) {
 
 	@Override
 	protected KafkaConsumerBackoffManager doCreateManager(ListenerContainerRegistry registry) {
-		KafkaConsumerBackoffManager kafkaConsumerBackoffManager = getKafkaConsumerBackoffManager(registry);
+		PartitionPausingBackoffManager kafkaConsumerBackoffManager = getKafkaConsumerBackoffManager(registry);
 		super.addApplicationListener(kafkaConsumerBackoffManager);
 		return kafkaConsumerBackoffManager;
 	}
@@ -142,10 +142,10 @@ protected final Clock getDefaultClock() {
 		return Clock.systemUTC();
 	}
 
-	private KafkaConsumerBackoffManager getKafkaConsumerBackoffManager(ListenerContainerRegistry registry) {
+	private PartitionPausingBackoffManager getKafkaConsumerBackoffManager(ListenerContainerRegistry registry) {
 		return this.timingAdjustmentEnabled
-			? new KafkaConsumerBackoffManager(registry, getOrCreateBackOffTimingAdjustmentManager(), this.clock)
-			: new KafkaConsumerBackoffManager(registry, this.clock);
+			? new PartitionPausingBackoffManager(registry, getOrCreateBackOffTimingAdjustmentManager(), this.clock)
+			: new PartitionPausingBackoffManager(registry, this.clock);
 	}
 
 	private KafkaConsumerTimingAdjuster getOrCreateBackOffTimingAdjustmentManager() {