GH-1210: Add nack capability to MANUAL AckModes

Resolves #1210

* Remove unused variable

* Fix race in tests

* Fix race in test

* Polishing - javadocs and docs

Author: garyrussell

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

@@ -70,6 +70,7 @@
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.kafka.support.LogIfLevelEnabled;
+import org.springframework.kafka.support.SeekUtils;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.TopicPartitionOffset.SeekPosition;
 import org.springframework.kafka.support.TransactionSupport;
@@ -523,6 +524,10 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private long lastAlertAt = this.lastReceive;
 
+		private long nackSleep = -1;
+
+		private int nackIndex;
+
 		private volatile boolean consumerPaused;
 
 		private volatile long lastPoll = System.currentTimeMillis();
@@ -1193,6 +1198,7 @@ private RuntimeException doInvokeBatchListener(final ConsumerRecords<K, V> recor
 
 		private void invokeBatchOnMessage(final ConsumerRecords<K, V> records, List<ConsumerRecord<K, V>> recordList,
 				@SuppressWarnings(RAW_TYPES) Producer producer) throws InterruptedException {
+
 			if (this.wantsFullRecords) {
 				this.batchListener.onMessage(records,
 						this.isAnyManualAck
@@ -1203,6 +1209,21 @@ private void invokeBatchOnMessage(final ConsumerRecords<K, V> records, List<Cons
 			else {
 				doInvokeBatchOnMessage(records, recordList);
 			}
+			List<ConsumerRecord<?, ?>> toSeek = null;
+			if (this.nackSleep >= 0) {
+				Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
+				int index = 0;
+				toSeek = new ArrayList<>();
+				while (iterator.hasNext()) {
+					ConsumerRecord<K, V> next = iterator.next();
+					if (index++ >= this.nackIndex) {
+						toSeek.add(next);
+					}
+					else {
+						this.acks.put(next);
+					}
+				}
+			}
 			if (!this.isAnyManualAck && !this.autoCommit) {
 				for (ConsumerRecord<K, V> record : getHighestOffsetRecords(records)) {
 					this.acks.put(record);
@@ -1211,10 +1232,18 @@ private void invokeBatchOnMessage(final ConsumerRecords<K, V> records, List<Cons
 					sendOffsetsToTransaction(producer);
 				}
 			}
+			if (this.nackSleep >= 0) {
+				if (!this.autoCommit) {
+					processCommits();
+				}
+				SeekUtils.doSeeks(toSeek, this.consumer, null, true, (rec, ex) -> false, this.logger);
+				this.nackSleep = -1;
+			}
 		}
 
 		private void doInvokeBatchOnMessage(final ConsumerRecords<K, V> records,
 				List<ConsumerRecord<K, V>> recordList) {
+
 			switch (this.listenerType) {
 				case ACKNOWLEDGING_CONSUMER_AWARE:
 					this.batchListener.onMessage(recordList,
@@ -1301,6 +1330,11 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 				finally {
 					TransactionSupport.clearTransactionIdSuffix();
 				}
+				if (this.nackSleep >= 0) {
+					handleNack(records, record);
+					break;
+				}
+
 			}
 		}
 
@@ -1341,9 +1375,35 @@ private void doInvokeWithRecords(final ConsumerRecords<K, V> records) {
 				final ConsumerRecord<K, V> record = iterator.next();
 				this.logger.trace(() -> "Processing " + record);
 				doInvokeRecordListener(record, null, iterator);
+				if (this.nackSleep >= 0) {
+					handleNack(records, record);
+					break;
+				}
 			}
 		}
 
+		private void handleNack(final ConsumerRecords<K, V> records, final ConsumerRecord<K, V> record) {
+			if (!this.autoCommit && !this.isRecordAck) {
+				processCommits();
+			}
+			List<ConsumerRecord<?, ?>> list = new ArrayList<>();
+			Iterator<ConsumerRecord<K, V>> iterator2 = records.iterator();
+			while (iterator2.hasNext()) {
+				ConsumerRecord<K, V> next = iterator2.next();
+				if (next.equals(record) || list.size() > 0) {
+					list.add(next);
+				}
+			}
+			SeekUtils.doSeeks(list, this.consumer, null, true, (rec, ex) -> false, this.logger);
+			try {
+				Thread.sleep(this.nackSleep);
+			}
+			catch (@SuppressWarnings("unused") InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+			this.nackSleep = -1;
+		}
+
 		/**
 		 * Actually invoke the listener.
 		 * @param record the record.
@@ -1399,7 +1459,9 @@ private void invokeOnMessage(final ConsumerRecord<K, V> record,
 				checkDeser(record, ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER);
 			}
 			doInvokeOnMessage(record);
-			ackCurrent(record, producer);
+			if (this.nackSleep < 0) {
+				ackCurrent(record, producer);
+			}
 		}
 
 		private void doInvokeOnMessage(final ConsumerRecord<K, V> recordArg) {
@@ -1829,11 +1891,17 @@ private final class ConsumerAcknowledgment implements Acknowledgment {
 
 			@Override
 			public void acknowledge() {
-				Assert.state(ListenerConsumer.this.isAnyManualAck,
-						"A manual ackmode is required for an acknowledging listener");
 				processAck(this.record);
 			}
 
+			@Override
+			public void nack(long sleep) {
+				Assert.state(Thread.currentThread().equals(ListenerConsumer.this.consumerThread),
+						"nack() can only be called on the consumer thread");
+				Assert.isTrue(sleep >= 0, "sleep cannot be negative");
+				ListenerConsumer.this.nackSleep = sleep;
+			}
+
 			@Override
 			public String toString() {
 				return "Acknowledgment for " + this.record;
@@ -1852,13 +1920,21 @@ private final class ConsumerBatchAcknowledgment implements Acknowledgment {
 
 			@Override
 			public void acknowledge() {
-				Assert.state(ListenerConsumer.this.isAnyManualAck,
-						"A manual ackmode is required for an acknowledging listener");
 				for (ConsumerRecord<K, V> record : getHighestOffsetRecords(this.records)) {
 					processAck(record);
 				}
 			}
 
+			@Override
+			public void nack(int index, long sleep) {
+				Assert.state(Thread.currentThread().equals(ListenerConsumer.this.consumerThread),
+						"nack() can only be called on the consumer thread");
+				Assert.isTrue(sleep >= 0, "sleep cannot be negative");
+				Assert.isTrue(index >= 0 && index < this.records.count(), "index out of bounds");
+				ListenerConsumer.this.nackIndex = index;
+				ListenerConsumer.this.nackSleep = sleep;
+			}
+
 			@Override
 			public String toString() {
 				return "Acknowledgment for " + this.records;

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

@@ -28,9 +28,34 @@
 public interface Acknowledgment {
 
 	/**
-	 * Invoked when the message for which the acknowledgment has been created has been processed.
-	 * Calling this method implies that all the previous messages in the partition have been processed already.
+	 * Invoked when the record or batch for which the acknowledgment has been created has
+	 * been processed. Calling this method implies that all the previous messages in the
+	 * partition have been processed already.
 	 */
 	void acknowledge();
 
+	/**
+	 * 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. Must be called on the consumer thread.
+	 * @param sleep the time to sleep.
+	 * @since 2.3
+	 */
+	default void nack(long 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. Must be called on
+	 * the consumer thread.
+	 * @param index the index of the failed record in the batch.
+	 * @param sleep the time to sleep.
+	 * @since 2.3
+	 */
+	default void nack(int index, long sleep) {
+		throw new UnsupportedOperationException("nack(index, sleep) is not supported by this Acknowledgment");
+	}
+
 }