GH-1353: Fix sleep with BatchListener nack

Resolves #1363

**cherry-pick to 2.3.x**

(cherry picked from commit cd6c2aa)

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -567,9 +567,10 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final boolean subBatchPerPartition = this.containerProperties.isSubBatchPerPartition();
 
-		private Map<TopicPartition, OffsetMetadata> definedPartitions;
+		private final Duration authorizationExceptionRetryInterval =
+				this.containerProperties.getAuthorizationExceptionRetryInterval();
 
-		private Duration authorizationExceptionRetryInterval = this.containerProperties.getAuthorizationExceptionRetryInterval();
+		private Map<TopicPartition, OffsetMetadata> definedPartitions;
 
 		private int count;
 
@@ -1438,7 +1439,7 @@ private void invokeBatchOnMessage(final ConsumerRecords<K, V> records, // NOSONA
 					processCommits();
 				}
 				SeekUtils.doSeeks(toSeek, this.consumer, null, true, (rec, ex) -> false, this.logger);
-				this.nackSleep = -1;
+				nackSleepAndReset();
 			}
 		}
 
@@ -1606,6 +1607,10 @@ private void handleNack(final ConsumerRecords<K, V> records, final ConsumerRecor
 				}
 			}
 			SeekUtils.doSeeks(list, this.consumer, null, true, (rec, ex) -> false, this.logger);
+			nackSleepAndReset();
+		}
+
+		private void nackSleepAndReset() {
 			try {
 				Thread.sleep(this.nackSleep);
 			}

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

@@ -91,6 +91,7 @@ public class ManualNackBatchTests {
 	@Test
 	public void discardRemainingRecordsFromPollAndSeek() throws Exception {
 		assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.config.replayTime).isBetween(50L, 30_000L);
 		assertThat(this.config.commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		this.registry.stop();
@@ -117,24 +118,27 @@ public void discardRemainingRecordsFromPollAndSeek() throws Exception {
 	@EnableKafka
 	public static class Config {
 
-		private final List<String> contents = new ArrayList<>();
+		final List<String> contents = new ArrayList<>();
 
-		private final CountDownLatch pollLatch = new CountDownLatch(3);
+		final CountDownLatch pollLatch = new CountDownLatch(3);
 
-		private final CountDownLatch deliveryLatch = new CountDownLatch(2);
+		final CountDownLatch deliveryLatch = new CountDownLatch(2);
 
-		private final CountDownLatch commitLatch = new CountDownLatch(2);
+		final CountDownLatch commitLatch = new CountDownLatch(2);
 
-		private final CountDownLatch closeLatch = new CountDownLatch(1);
+		final CountDownLatch closeLatch = new CountDownLatch(1);
 
-		private final AtomicBoolean fail = new AtomicBoolean(true);
+		final AtomicBoolean fail = new AtomicBoolean(true);
+
+		volatile long replayTime;
 
 		@KafkaListener(id = CONTAINER_ID, topics = "foo")
 		public void foo(List<String> in, Acknowledgment ack) {
 			contents.addAll(in);
+			this.replayTime = System.currentTimeMillis() - this.replayTime;
 			this.deliveryLatch.countDown();
 			if (this.fail.getAndSet(false)) {
-				ack.nack(3, 0);
+				ack.nack(3, 50);
 			}
 			else {
 				ack.acknowledge();