GH-2410: Disallow nack() with Out of Order Commits

Resolves #2410

`nack()` cannot be used with out of order commits
- the contract means commit all previous offsets and redeliver remaining.
- the appliation might nack multiple records.

**cherry-pick to 2.9.x, 2.8.x**

# Conflicts:
#	spring-kafka-docs/src/main/asciidoc/kafka.adoc

Author: garyrussell

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

@@ -1283,6 +1283,8 @@ NOTE: If you want to commit a partial batch, using `nack()`, When using transact
 
 IMPORTANT: `nack()` can only be called on the consumer thread that invokes your listener.
 
+IMPORTANT: `nack()` is not allowed when using <<ooo-commits, Out of Order Commits>>.
+
 With a record listener, when `nack()` is called, any pending offsets are committed, the remaining records from the last poll are discarded, and seeks are performed on their partitions so that the failed record and unprocessed records are redelivered on the next `poll()`.
 The consumer can be paused before redelivery, by setting the `sleep` argument.
 This is similar functionality to throwing an exception when the container is configured with a `DefaultErrorHandler`.

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

@@ -3343,14 +3343,10 @@ public void acknowledge() {
 			public void nack(Duration sleep) {
 				Assert.state(Thread.currentThread().equals(ListenerConsumer.this.consumerThread),
 						"nack() can only be called on the consumer thread");
+				Assert.state(!ListenerConsumer.this.containerProperties.isAsyncAcks(),
+						"nack() is not supported with out-of-order commits (asyncAcks=true)");
 				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());
-						ListenerConsumer.this.deferredOffsets.forEach((part, recs) -> recs.clear());
-					}
-				}
 			}
 
 			@Override
@@ -3391,16 +3387,12 @@ public void acknowledge() {
 			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.state(!ListenerConsumer.this.containerProperties.isAsyncAcks(),
+						"nack() is not supported with out-of-order commits (asyncAcks=true)");
 				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 = sleep.toMillis();
-				synchronized (ListenerConsumer.this) {
-					if (ListenerConsumer.this.offsetsInThisBatch != null) {
-						ListenerConsumer.this.offsetsInThisBatch.forEach((part, recs) -> recs.clear());
-						ListenerConsumer.this.deferredOffsets.forEach((part, recs) -> recs.clear());
-					}
-				}
 				int i = 0;
 				List<ConsumerRecord<K, V>> toAck = new LinkedList<>();
 				for (ConsumerRecord<K, V> record : this.records) {

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

@@ -689,7 +689,16 @@ private void testInOrderAck(AckMode ackMode) throws Exception {
 		containerProps.setAsyncAcks(true);
 		final CountDownLatch latch = new CountDownLatch(4);
 		final List<Acknowledgment> acks = new ArrayList<>();
+		final AtomicReference<IllegalStateException> illegal = new AtomicReference<>();
 		AcknowledgingMessageListener<Integer, String> messageListener = (data, ack) -> {
+			if (latch.getCount() == 4) {
+				try {
+					ack.nack(Duration.ofSeconds(1));
+				}
+				catch (IllegalStateException ex) {
+					illegal.set(ex);
+				}
+			}
 			latch.countDown();
 			acks.add(ack);
 			if (latch.getCount() == 0) {
@@ -720,6 +729,7 @@ private void testInOrderAck(AckMode ackMode) throws Exception {
 		verify(consumer).commitSync(Map.of(new TopicPartition("foo", 0), new OffsetAndMetadata(4L)),
 				Duration.ofMinutes(1));
 		container.stop();
+		assertThat(illegal.get()).isNotNull();
 	}
 
 	@Test