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

Author: garyrussell

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

@@ -1219,6 +1219,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`.
@@ -3783,7 +3785,7 @@ If you go with this approach, then you need to set this producer interceptor on
 Following is an example using the same `MyProducerInterceptor` from above, but changed to not use the internal config property.
 
 ====
-[source]
+[source, java]
 ----
 public class MyProducerInterceptor implements ProducerInterceptor<String, String> {
 

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

@@ -3382,14 +3382,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
@@ -3429,16 +3425,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

@@ -692,7 +692,16 @@ 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) {
@@ -723,6 +732,7 @@ 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();
 	}
 
 	private static Stream<Arguments> testInOrderAckPauseUntilAckedParamters() {