GH-566: Ack Concurrency Issue

Fixes #566

If `Acknowledgment.acknowledge()` is called on a foreign thread, there is a concurrency
problem with the `offsets` field; the consumer thread might `clear()` unprocessed acks.

Furthermore, `AckMode.MANUAL_IMMEDIATE` cannot use the `Consumer` object if the ack
is called on a foreign thread - the `Consumer` is not thread-safe.

- Revert `offsets` to simple `HashMap`s
- Only reference `offsets` on the consumer thread
- enqueue foreign acks into the `acks` queue (even "immediate" acks)

**Cherry pick to 2.0.x (and 1.3.x, fixing the lambda in the test and the diamond operators).**

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

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
#	spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

Author: garyrussell

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

@@ -48,6 +48,7 @@
 import org.apache.kafka.common.errors.WakeupException;
 
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaResourceHolder;
 import org.springframework.kafka.core.ProducerFactoryUtils;
@@ -293,7 +294,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final Consumer<K, V> consumer;
 
-		private final Map<String, Map<Integer, Long>> offsets = new HashMap<>();
+		private final Map<String, Map<Integer, Long>> offsets = new HashMap<String, Map<Integer, Long>>();
 
 		private final MessageListener<K, V> listener;
 
@@ -348,6 +349,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private volatile Collection<TopicPartition> assignedPartitions;
 
+		private volatile Thread consumerThread;
+
 		private int count;
 
 		private long last = System.currentTimeMillis();
@@ -600,6 +603,7 @@ public boolean isLongLived() {
 
 		@Override
 		public void run() {
+			this.consumerThread = Thread.currentThread();
 			if (this.theListener instanceof ConsumerSeekAware) {
 				((ConsumerSeekAware) this.theListener).registerSeekCallback(this);
 			}
@@ -709,16 +713,27 @@ record = this.acks.poll();
 		}
 
 		private void processAck(ConsumerRecord<K, V> record) {
-			if (this.isManualImmediateAck) {
+			if (!Thread.currentThread().equals(this.consumerThread)) {
 				try {
-					ackImmediate(record);
+					this.acks.put(record);
 				}
-				catch (WakeupException e) {
-					// ignore - not polling
+				catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+					throw new KafkaException("Interrupted while storing ack", e);
 				}
 			}
 			else {
-				addOffset(record);
+				if (this.isManualImmediateAck) {
+					try {
+						ackImmediate(record);
+					}
+					catch (WakeupException e) {
+						// ignore - not polling
+					}
+				}
+				else {
+					addOffset(record);
+				}
 			}
 		}
 

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

@@ -22,6 +22,7 @@
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyLong;
 import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.eq;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -308,6 +309,77 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
 		container.stop();
 	}
 
+	@Test
+	public void testRecordAckMockForeignThread() throws Exception {
+		testRecordAckMockForeignThreadGuts(AckMode.MANUAL);
+	}
+
+	@Test
+	public void testRecordAckMockForeignThreadImmediate() throws Exception {
+		testRecordAckMockForeignThreadGuts(AckMode.MANUAL_IMMEDIATE);
+	}
+
+	@SuppressWarnings("unchecked")
+	private void testRecordAckMockForeignThreadGuts(AckMode ackMode) throws Exception {
+		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
+		Consumer<Integer, String> consumer = mock(Consumer.class);
+		given(cf.createConsumer(anyString(), eq("clientId"))).willReturn(consumer);
+		final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new HashMap<>();
+		records.put(new TopicPartition("foo", 0), Arrays.asList(
+				new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
+				new ConsumerRecord<>("foo", 0, 1L, 1, "bar")));
+		ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);
+		given(consumer.poll(anyLong())).willAnswer(i -> {
+			Thread.sleep(50);
+			return consumerRecords;
+		});
+		TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] {
+				new TopicPartitionInitialOffset("foo", 0) };
+		ContainerProperties containerProps = new ContainerProperties(topicPartition);
+		containerProps.setAckMode(ackMode);
+		final CountDownLatch latch = new CountDownLatch(2);
+		final List<Acknowledgment> acks = new ArrayList<>();
+		final AtomicReference<Thread> consumerThread = new AtomicReference<>();
+		AcknowledgingMessageListener<Integer, String> messageListener = spy(
+				new AcknowledgingMessageListener<Integer, String>() {
+
+					@Override
+					public void onMessage(ConsumerRecord<Integer, String> data, Acknowledgment acknowledgment) {
+						acks.add(acknowledgment);
+						consumerThread.set(Thread.currentThread());
+						latch.countDown();
+						if (latch.getCount() == 0) {
+							records.clear();
+						}
+					}
+
+				});
+
+		final CountDownLatch commitLatch = new CountDownLatch(1);
+		final AtomicReference<Thread> commitThread = new AtomicReference<>();
+		willAnswer(i -> {
+					commitThread.set(Thread.currentThread());
+					commitLatch.countDown();
+					return null;
+				}
+		).given(consumer).commitSync(any(Map.class));
+
+		containerProps.setMessageListener(messageListener);
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.setClientIdSuffix("clientId");
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		acks.get(1).acknowledge();
+		assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		InOrder inOrder = inOrder(messageListener, consumer);
+		inOrder.verify(consumer).poll(1000);
+		inOrder.verify(messageListener, times(2)).onMessage(any(ConsumerRecord.class), any(Acknowledgment.class));
+		inOrder.verify(consumer).commitSync(any(Map.class));
+		container.stop();
+		assertThat(commitThread.get()).isSameAs(consumerThread.get());
+	}
+
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testBrokerDownEvent() throws Exception {