AckMode.MANUAL_IMMEDIATE - wake Consumer

When processing manual acks on a "foreign" thread, the commits are
not made until the consumer wakes from his `poll()`.

For `AckMode.MANUAL_IMMEDIATE`, we should wake the consumer so the
offset(s) are committed as soon as possible.

**cherry-pick to 2.2.x**

Author: garyrussell

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

@@ -700,7 +700,7 @@ public void run() {
 					pollAndInvoke();
 				}
 				catch (@SuppressWarnings(UNUSED) WakeupException e) {
-					// Ignore, we're stopping
+					// Ignore, we're stopping or applying immediate foreign acks
 				}
 				catch (NoOffsetForPartitionException nofpe) {
 					this.fatalError = true;
@@ -893,6 +893,9 @@ private void processAck(ConsumerRecord<K, V> record) {
 			if (!Thread.currentThread().equals(this.consumerThread)) {
 				try {
 					this.acks.put(record);
+					if (this.isManualImmediateAck) {
+						this.consumer.wakeup();
+					}
 				}
 				catch (InterruptedException e) {
 					Thread.currentThread().interrupt();

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

@@ -608,15 +608,25 @@ private void testRecordAckMockForeignThreadGuts(AckMode ackMode) throws Exceptio
 				new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
 				new ConsumerRecord<>("foo", 0, 1L, 1, "bar")));
 		ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);
+		long sleepFor = ackMode.equals(AckMode.MANUAL_IMMEDIATE) ? 20_000 : 50;
+		AtomicBoolean first = new AtomicBoolean(true);
 		given(consumer.poll(any(Duration.class))).willAnswer(i -> {
-			Thread.sleep(50);
+			if (!first.getAndSet(false)) {
+				try {
+					Thread.sleep(sleepFor);
+				}
+				catch (@SuppressWarnings("unused") InterruptedException ex) {
+					throw new WakeupException();
+				}
+			}
 			return consumerRecords;
 		});
 		TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] {
 				new TopicPartitionInitialOffset("foo", 0) };
 		ContainerProperties containerProps = new ContainerProperties(topicPartition);
 		containerProps.setGroupId("grp");
 		containerProps.setAckMode(ackMode);
+		containerProps.setMissingTopicsFatal(false);
 		final CountDownLatch latch = new CountDownLatch(2);
 		final List<Acknowledgment> acks = new ArrayList<>();
 		final AtomicReference<Thread> consumerThread = new AtomicReference<>();
@@ -634,6 +644,10 @@ public void onMessage(ConsumerRecord<Integer, String> data, Acknowledgment ackno
 					}
 
 				});
+		willAnswer(inv -> {
+			consumerThread.get().interrupt();
+			return null;
+		}).given(consumer).wakeup();
 
 		final CountDownLatch commitLatch = new CountDownLatch(1);
 		final AtomicReference<Thread> commitThread = new AtomicReference<>();
@@ -650,6 +664,7 @@ public void onMessage(ConsumerRecord<Integer, String> data, Acknowledgment ackno
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.start();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		long t1 = System.currentTimeMillis();
 		acks.get(1).acknowledge();
 		assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		InOrder inOrder = inOrder(messageListener, consumer);
@@ -658,6 +673,7 @@ public void onMessage(ConsumerRecord<Integer, String> data, Acknowledgment ackno
 		inOrder.verify(consumer).commitSync(any(Map.class));
 		container.stop();
 		assertThat(commitThread.get()).isSameAs(consumerThread.get());
+		assertThat(System.currentTimeMillis() - t1).isLessThan(15_000);
 	}
 
 	@SuppressWarnings("unchecked")