Pause/Resume and Acknowledgment Improvements

- wake consumer when pausing/resuming if currently blocked in `poll()`.
- use `stoppableSleep` in `nackSleepAndReset`

Author: garyrussell

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

@@ -292,6 +292,24 @@ public boolean isInExpectedState() {
 		return isRunning() || isStoppedNormally();
 	}
 
+	@Override
+	public void pause() {
+		super.pause();
+		KafkaMessageListenerContainer<K, V>.ListenerConsumer consumer = this.listenerConsumer;
+		if (consumer != null) {
+			consumer.wakeIfNecessary();
+		}
+	}
+
+	@Override
+	public void resume() {
+		super.resume();
+		KafkaMessageListenerContainer<K, V>.ListenerConsumer consumer = this.listenerConsumer;
+		if (consumer != null) {
+			this.listenerConsumer.wakeIfNecessary();
+		}
+	}
+
 	@Override
 	public Map<String, Map<MetricName, ? extends Metric>> metrics() {
 		ListenerConsumer listenerConsumerForMetrics = this.listenerConsumer;
@@ -369,7 +387,7 @@ protected void doStop(final Runnable callback, boolean normal) {
 		if (isRunning()) {
 			this.listenerConsumerFuture.addCallback(new StopCallback(callback));
 			setRunning(false);
-			this.listenerConsumer.wakeIfNecessary();
+			this.listenerConsumer.wakeIfNecessaryForStop();
 			setStoppedNormally(normal);
 		}
 	}
@@ -1303,7 +1321,7 @@ protected void pollAndInvoke() {
 			ConsumerRecords<K, V> records = doPoll();
 			if (!this.polling.compareAndSet(true, false) && records != null) {
 				/*
-				 * There is a small race condition where wakeIfNecessary was called between
+				 * There is a small race condition where wakeIfNecessaryForStop was called between
 				 * exiting the poll and before we reset the boolean.
 				 */
 				if (records.count() > 0) {
@@ -1521,12 +1539,18 @@ private void checkRebalanceCommits() {
 			}
 		}
 
-		void wakeIfNecessary() {
+		void wakeIfNecessaryForStop() {
 			if (this.polling.getAndSet(false)) {
 				this.consumer.wakeup();
 			}
 		}
 
+		void wakeIfNecessary() {
+			if (this.polling.get()) {
+				this.consumer.wakeup();
+			}
+		}
+
 		private void debugRecords(@Nullable ConsumerRecords<K, V> records) {
 			if (records != null) {
 				this.logger.debug(() -> "Received: " + records.count() + " records");
@@ -2427,7 +2451,7 @@ private void handleNack(final ConsumerRecords<K, V> records, final ConsumerRecor
 
 		private void nackSleepAndReset() {
 			try {
-				Thread.sleep(this.nackSleep);
+				ListenerUtils.stoppableSleep(KafkaMessageListenerContainer.this.thisOrParentContainer, this.nackSleep);
 			}
 			catch (@SuppressWarnings(UNUSED) InterruptedException e) {
 				Thread.currentThread().interrupt();