Add Acknowledgment.nack() variants accepting Duration

To prevent confusion what unit should be used for "long sleep" arguments.
As the old and new methods in the Acknowledgment interface have default
implementations, the change itself is backward compatible.

The old methods are marked as deprecated and intended to be removed in the future.

Author: szpak

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

@@ -3174,11 +3174,11 @@ public void acknowledge() {
 			}
 
 			@Override
-			public void nack(long sleepMillis) {
+			public void nack(Duration sleep) {
 				Assert.state(Thread.currentThread().equals(ListenerConsumer.this.consumerThread),
 						"nack() can only be called on the consumer thread");
-				Assert.isTrue(sleepMillis >= 0, "sleepMillis cannot be negative");
-				ListenerConsumer.this.nackSleepDurationMillis = sleepMillis;
+				Assert.isTrue(!sleep.isNegative(), "sleep cannot be negative");
+				ListenerConsumer.this.nackSleepDurationMillis = sleep.toMillis();
 				synchronized (ListenerConsumer.this) {
 					if (ListenerConsumer.this.offsetsInThisBatch != null) {
 						ListenerConsumer.this.offsetsInThisBatch.forEach((part, recs) -> recs.clear());
@@ -3221,13 +3221,13 @@ public void acknowledge() {
 			}
 
 			@Override
-			public void nack(int index, long sleepMillis) {
+			public void nack(int index, Duration sleep) {
 				Assert.state(Thread.currentThread().equals(ListenerConsumer.this.consumerThread),
 						"nack() can only be called on the consumer thread");
-				Assert.isTrue(sleepMillis >= 0, "sleepMillis cannot be negative");
+				Assert.isTrue(!sleep.isNegative(), "sleep cannot be negative");
 				Assert.isTrue(index >= 0 && index < this.records.count(), "index out of bounds");
 				ListenerConsumer.this.nackIndex = index;
-				ListenerConsumer.this.nackSleepDurationMillis = sleepMillis;
+				ListenerConsumer.this.nackSleepDurationMillis = sleep.toMillis();
 				synchronized (ListenerConsumer.this) {
 					if (ListenerConsumer.this.offsetsInThisBatch != null) {
 						ListenerConsumer.this.offsetsInThisBatch.forEach((part, recs) -> recs.clear());

spring-kafka/src/main/java/org/springframework/kafka/support/Acknowledgment.java

@@ -16,6 +16,8 @@
 
 package org.springframework.kafka.support;
 
+import java.time.Duration;
+
 /**
  * Handle for acknowledging the processing of a
  * {@link org.apache.kafka.clients.consumer.ConsumerRecord}. Recipients can store the
@@ -42,8 +44,23 @@ public interface Acknowledgment {
 	 * @param sleepMillis the time to sleep in milliseconds; the actual sleep time will be larger
 	 * of this value and the container's {@code maxPollInterval}, which defaults to 5 seconds.
 	 * @since 2.3
+	 * @deprecated in favor of {@link #nack(Duration)}
 	 */
+	@Deprecated(since = "2.8.7", forRemoval = true)
 	default void nack(long sleepMillis) {
+		nack(Duration.ofMillis(sleepMillis));
+	}
+
+	/**
+	 * Negatively acknowledge the current record - discard remaining records from the poll
+	 * and re-seek all partitions so that this record will be redelivered after the sleep
+	 * duration. Must be called on the consumer thread.
+	 * <p>
+	 * @param sleep the duration to sleep; the actual sleep time will be larger of this value
+	 * and the container's {@code maxPollInterval}, which defaults to 5 seconds.
+	 * @since 2.8.7
+	 */
+	default void nack(Duration sleep) {
 		throw new UnsupportedOperationException("nack(sleep) is not supported by this Acknowledgment");
 	}
 
@@ -57,8 +74,25 @@ default void nack(long sleepMillis) {
 	 * @param sleepMillis the time to sleep in milliseconds; the actual sleep time will be larger
 	 * of this value and the container's {@code maxPollInterval}, which defaults to 5 seconds.
 	 * @since 2.3
+	 * @deprecated in favor of {@link #nack(int, Duration)}
 	 */
+	@Deprecated(since = "2.8.7", forRemoval = true)
 	default void nack(int index, long sleepMillis) {
+		nack(index, Duration.ofMillis(sleepMillis));
+	}
+
+	/**
+	 * Negatively acknowledge the record at an index in a batch - commit the offset(s) of
+	 * records before the index and re-seek the partitions so that the record at the index
+	 * and subsequent records will be redelivered after the sleep duration.
+	 * Must be called on the consumer thread.
+	 * <p>
+	 * @param index the index of the failed record in the batch.
+	 * @param sleep the duration to sleep; the actual sleep time will be larger of this value
+	 * and the container's {@code maxPollInterval}, which defaults to 5 seconds.
+	 * @since 2.8.7
+	 */
+	default void nack(int index, Duration sleep) {
 		throw new UnsupportedOperationException("nack(index, sleep) is not supported by this Acknowledgment");
 	}
 

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

@@ -146,7 +146,7 @@ public void foo(List<String> in, Acknowledgment ack) {
 			this.replayTime = System.currentTimeMillis() - this.replayTime;
 			this.deliveryLatch.countDown();
 			if (this.fail.getAndSet(false)) {
-				ack.nack(3, 50);
+				ack.nack(3, Duration.ofMillis(50));
 			}
 			else {
 				ack.acknowledge();

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

@@ -150,7 +150,7 @@ public void foo(List<String> in, Acknowledgment ack) {
 			this.replayTime = System.currentTimeMillis() - this.replayTime;
 			this.deliveryLatch.countDown();
 			if (++this.count == 1) { // part 1, offset 1, first time
-				ack.nack(3, 50);
+				ack.nack(3, Duration.ofMillis(50));
 			}
 			else {
 				ack.acknowledge();

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

@@ -125,7 +125,7 @@ public void foo(String in, Acknowledgment ack) {
 			}
 			this.deliveryLatch.countDown();
 			if (++this.count == 4) { // part 1, offset 1, first time
-				ack.nack(50);
+				ack.nack(Duration.ofMillis(50));
 			}
 			else {
 				ack.acknowledge();

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

@@ -142,7 +142,7 @@ public void foo(String in, Acknowledgment ack) {
 			}
 			this.deliveryLatch.countDown();
 			if (++this.count == 4) { // part 1, offset 1, first time
-				ack.nack(50);
+				ack.nack(Duration.ofMillis(50));
 			}
 			else {
 				ack.acknowledge();

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

@@ -136,7 +136,7 @@ public void foo(String in, Acknowledgment ack) {
 			++this.count;
 			if (this.contents.size() == 1 || this.count == 5 || this.count == 8) {
 				// first, last record or part 1, offset 1, first time
-				ack.nack(0);
+				ack.nack(Duration.ofMillis(0));
 			}
 			else {
 				ack.acknowledge();