Gh 1716 - Refactoring and Test Coverage (#1734)

* Fixes GH-1716.

* GH-1716 - Fixing Copy/Paste artifact.

* GH-1716 - Refactorings and improved test coverage.

* Checkstyle adjustments.

* Code review adjustments.

* Adjust test names and exception inheritance

Author: tomazfernandes

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

@@ -23,6 +23,27 @@ IMPORTANT: You can set the `AckMode` mode you prefer, but `RECORD` is suggested.
 
 IMPORTANT: At this time this functionality doesn't support class level `@KafkaListener` annotations
 
+=== Back Off Delay Precision
+
+==== Overview and Guarantees
+
+All message processing and backing off is handled by the consumer thread, and, as such, delay precision is guaranteed on a best-effort basis.
+If one message's processing takes longer than the next message's back off period for that consumer, the next message's delay will be higher than expected.
+Also, for short delays (about 1s or less), the maintenance work the thread has to do, such as committing offsets, may delay the message processing execution.
+The precision can also be affected if the retry topic's consumer is handling more than one partition, because we rely on waking up the consumer from polling and having full pollTimeouts to make timing adjustments.
+
+That being said, for consumers handling a single partition the message's processing should happen under 100ms after it's exact due time for most situations.
+
+IMPORTANT: It is guaranteed that a message will never be processed before its due time.
+
+==== Tuning the Delay Precision
+
+The message's processing delay precision relies on two `ContainerProperties`: `ContainerProperties.pollTimeout` and `ContainerProperties.idlePartitionEventInterval`.
+Both properties will be automatically set in the retry topic and dlt's `ListenerContainerFactory` to one quarter of the smallest delay value for that topic, with a minimum value of 250ms and a maximum value of 5000ms.
+These values will only be set if the property has its default values - if you change either value yourself your change will not be overridden.
+This way you can tune the precision and performance for the retry topics if you need to.
+
+NOTE: You can have separate `ListenerContainerFactory` instances for the main and retry topics - this way you can have different settings to better suit your needs, such as having a higher polling timeout setting for the main topics and a lower one for the retry topics.
 
 === Configuration
 

spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopic.java

@@ -174,7 +174,7 @@
 	TopicSuffixingStrategy topicSuffixingStrategy() default TopicSuffixingStrategy.SUFFIX_WITH_DELAY_VALUE;
 
 	/**
-	 * Whether or not to redeliver to the DLT if delivery fails.
+	 * Whether or not create a DLT, and redeliver to the DLT if delivery fails or just give up.
 	 * @return the dlt strategy.
 	 */
 	DltStrategy dltStrategy() default DltStrategy.ALWAYS_RETRY_ON_ERROR;

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

@@ -18,15 +18,20 @@
 
 import java.time.Clock;
 import java.time.Instant;
-import java.time.temporal.ChronoUnit;
 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.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;
 
 /**
  *
@@ -47,87 +52,163 @@
  */
 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 Map<TopicPartition, Context> backOffTimes;
+	private final Map<TopicPartition, Context> backOffContexts;
 
 	private final Clock clock;
 
+	private final TaskExecutor taskExecutor;
+
+	private final Sleeper sleeper;
+
 	public KafkaConsumerBackoffManager(ListenerContainerRegistry registry,
-									@Qualifier(INTERNAL_BACKOFF_CLOCK_BEAN_NAME) Clock clock) {
+									@Qualifier(INTERNAL_BACKOFF_CLOCK_BEAN_NAME) Clock clock,
+									TaskExecutor taskExecutor,
+									Sleeper sleeper) {
 
 		this.registry = registry;
 		this.clock = clock;
-		this.backOffTimes = new HashMap<>();
+		this.taskExecutor = taskExecutor;
+		this.sleeper = sleeper;
+		this.backOffContexts = new HashMap<>();
 	}
 
 	/**
 	 * Backs off if the current time is before the dueTimestamp provided
 	 * in the {@link Context} object.
-	 * @param context the state that will be used for backing off.
+	 * @param context the back off context for this execution.
 	 */
 	public void maybeBackoff(Context context) {
-		long backoffTime = ChronoUnit.MILLIS.between(Instant.now(this.clock),
-				Instant.ofEpochMilli(context.dueTimestamp));
+		long backoffTime = context.dueTimestamp - getCurrentMillisFromClock();
 		if (backoffTime > 0) {
 			pauseConsumptionAndThrow(context, backoffTime);
 		}
 	}
 
-	private void pauseConsumptionAndThrow(Context context, Long timeToSleep) throws KafkaBackoffException {
+	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(), timeToSleep),
+				context.topicPartition.topic(), backOffTime),
 				topicPartition, context.listenerId, context.dueTimestamp);
 	}
 
 	@Override
 	public void onApplicationEvent(ListenerContainerPartitionIdleEvent partitionIdleEvent) {
-		Context context = getBackoff(partitionIdleEvent.getTopicPartition());
-		if (context == null || isNotDue(context.dueTimestamp)) {
+		logger.debug(() -> String.format("partitionIdleEvent received at %s. Partition: %s",
+				getCurrentMillisFromClock(), partitionIdleEvent.getTopicPartition()));
+
+		Context backOffContext = getBackOffContext(partitionIdleEvent.getTopicPartition());
+
+		if (backOffContext == null) {
 			return;
 		}
+		maybeResumeConsumption(backOffContext);
+	}
+
+	private long getCurrentMillisFromClock() {
+		return Instant.now(this.clock).toEpochMilli();
+	}
+
+	private void maybeResumeConsumption(Context context) {
+		long now = getCurrentMillisFromClock();
+		long timeUntilDue = context.dueTimestamp - now;
+		long pollTimeout = getListenerContainerFromContext(context)
+				.getContainerProperties()
+				.getPollTimeout();
+		boolean isDue = timeUntilDue <= pollTimeout;
+
+		if (maybeApplyTimingCorrection(context, pollTimeout, timeUntilDue) || isDue) {
+			resumePartition(context);
+		}
+		else {
+			logger.debug(() -> String.format("TopicPartition %s not due. DueTimestamp: %s Now: %s ",
+					context.topicPartition, context.dueTimestamp, now));
+		}
+	}
+
+	private void resumePartition(Context context) {
 		MessageListenerContainer container = getListenerContainerFromContext(context);
+		logger.debug(() -> "Resuming partition at " + getCurrentMillisFromClock());
 		container.resumePartition(context.topicPartition);
 		removeBackoff(context.topicPartition);
 	}
 
-	private boolean isNotDue(long dueTimestamp) {
-		return Instant.now(this.clock).isBefore(Instant.ofEpochMilli(dueTimestamp));
+	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.interrupted();
+			throw new IllegalStateException("Interrupted waking up consumer while applying correction " +
+					"for TopicPartition " + context.topicPartition, e);
+		}
+		catch (Throwable e) {
+			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);
 	}
 
 	protected void addBackoff(Context context, TopicPartition topicPartition) {
-		synchronized (this.backOffTimes) {
-			this.backOffTimes.put(topicPartition, context);
+		synchronized (this.backOffContexts) {
+			this.backOffContexts.put(topicPartition, context);
 		}
 	}
 
-	protected Context getBackoff(TopicPartition topicPartition) {
-		synchronized (this.backOffTimes) {
-			return this.backOffTimes.get(topicPartition);
+	protected Context getBackOffContext(TopicPartition topicPartition) {
+		synchronized (this.backOffContexts) {
+			return this.backOffContexts.get(topicPartition);
 		}
 	}
 
 	protected void removeBackoff(TopicPartition topicPartition) {
-		synchronized (this.backOffTimes) {
-			this.backOffTimes.remove(topicPartition);
+		synchronized (this.backOffContexts) {
+			this.backOffContexts.remove(topicPartition);
 		}
 	}
 
-	public Context createContext(long dueTimestamp, String listenerId, TopicPartition topicPartition) {
-		return new Context(dueTimestamp, listenerId, topicPartition);
+	public Context createContext(long dueTimestamp, String listenerId, TopicPartition topicPartition,
+								@Nullable Consumer<?, ?> consumerForTimingCorrection) {
+		return new Context(dueTimestamp, topicPartition, listenerId, consumerForTimingCorrection);
 	}
 
 	/**
@@ -140,22 +221,29 @@ public static class Context {
 		 * The time after which the message should be processed,
 		 * in milliseconds since epoch.
 		 */
-		final long dueTimestamp; // NOSONAR
+		private final long dueTimestamp; // NOSONAR
 
 		/**
 		 * The id for the listener that should be paused.
 		 */
-		final String listenerId; // NOSONAR
+		private final String listenerId; // NOSONAR
 
 		/**
 		 * The topic that contains the partition to be paused.
 		 */
-		final TopicPartition topicPartition; // NOSONAR
+		private final TopicPartition topicPartition; // NOSONAR
+
+		/**
+		 * The consumer of the message, if present.
+		 */
+		private final Consumer<?, ?> consumerForTimingCorrection; // NOSONAR
 
-		Context(long dueTimestamp, String listenerId, TopicPartition topicPartition) {
+		Context(long dueTimestamp, TopicPartition topicPartition, String listenerId,
+						@Nullable Consumer<?, ?> consumerForTimingCorrection) {
 			this.dueTimestamp = dueTimestamp;
 			this.listenerId = listenerId;
 			this.topicPartition = topicPartition;
+			this.consumerForTimingCorrection = consumerForTimingCorrection;
 		}
 	}
 }

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

@@ -1260,6 +1260,7 @@ protected void pollAndInvoke() {
 			resumeConsumerIfNeccessary();
 			resumePartitionsIfNecessary();
 			debugRecords(records);
+
 			if (records != null && records.count() > 0) {
 				savePositionsIfNeeded(records);
 				notIdle();
@@ -1268,6 +1269,9 @@ protected void pollAndInvoke() {
 			}
 			else {
 				checkIdle();
+			}
+			if (records == null || records.count() == 0
+					|| records.partitions().size() < this.consumer.assignment().size()) {
 				checkIdlePartitions();
 			}
 		}
@@ -2212,7 +2216,12 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> recor
 					commitOffsetsIfNeeded(record);
 				}
 				catch (KafkaException ke) {
-					ke.selfLog(ERROR_HANDLER_THREW_AN_EXCEPTION, this.logger);
+					if (ke.contains(KafkaBackoffException.class)) {
+						this.logger.warn(ke.getMessage());
+					}
+					else {
+						ke.selfLog(ERROR_HANDLER_THREW_AN_EXCEPTION, this.logger);
+					}
 					return ke;
 				}
 				catch (RuntimeException ee) {

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

@@ -0,0 +1,50 @@
+/*
+ * 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 java.time.Instant;
+
+import org.springframework.kafka.KafkaException;
+
+/**
+ * A {@link KafkaException} that records the timestamp
+ * of when it was thrown.
+ *
+ * @author Tomaz Fernandes
+ * @since 2.7
+ */
+public class TimestampedException extends KafkaException {
+
+	private static final long serialVersionUID = -2544217643924234282L;
+
+	private final long timestamp;
+
+	public TimestampedException(Exception ex, Clock clock) {
+		super(ex.getMessage(), ex);
+		this.timestamp = Instant.now(clock).toEpochMilli();
+	}
+
+	public TimestampedException(Exception ex) {
+		super(ex.getMessage(), ex);
+		this.timestamp = Instant.now(Clock.systemDefaultZone()).toEpochMilli();
+	}
+
+	public long getTimestamp() {
+		return this.timestamp;
+	}
+}