Add Acknowledgment.nack() variants accepting Duration (#2281)

* 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.

* Remove deprecated Acknowledgment.nack(long) method

nack(Duration) should be used instead.

Author: szpak

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

@@ -3182,11 +3182,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());
@@ -3229,13 +3229,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());

Diff for: 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
@@ -37,28 +39,28 @@ public interface Acknowledgment {
 	/**
 	 * 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
-	 * time (in milliseconds). Must be called on the consumer thread.
+	 * duration. Must be called on the consumer thread.
 	 * <p>
-	 * @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
+	 * @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(long sleepMillis) {
+	default void nack(Duration sleep) {
 		throw new UnsupportedOperationException("nack(sleep) is not supported by this Acknowledgment");
 	}
 
 	/**
 	 * 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 time (in milliseconds).
+	 * 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 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
+	 * @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, long sleepMillis) {
+	default void nack(int index, Duration sleep) {
 		throw new UnsupportedOperationException("nack(index, sleep) is not supported by this Acknowledgment");
 	}
 

Diff for: 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();

Diff for: 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();

Diff for: 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();

Diff for: 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();

Diff for: 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();