GH-511: Committing offset immediately on error

Fixes: #511

**Cherry-pick to 2.0.x & master**

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

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

Author: Tom van den Berge

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

@@ -88,6 +88,7 @@
  * @author Loic Talhouarne
  * @author Vladimir Tsanev
  * @author Yang Qiju
+ * @author Tom van den Berge
  */
 public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {
 
@@ -968,7 +969,20 @@ else if (!this.isAnyManualAck && !this.autoCommit) {
 			}
 			catch (RuntimeException e) {
 				if (this.containerProperties.isAckOnError() && !this.autoCommit && producer == null) {
-					this.acks.add(record);
+					if (this.isRecordAck) {
+						Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =
+								Collections.singletonMap(new TopicPartition(record.topic(), record.partition()),
+										new OffsetAndMetadata(record.offset() + 1));
+						if (this.containerProperties.isSyncCommits()) {
+							this.consumer.commitSync(offsetsToCommit);
+						}
+						else {
+							this.consumer.commitAsync(offsetsToCommit, this.commitCallback);
+						}
+					}
+					else if (!this.isAnyManualAck) {
+						this.acks.add(record);
+					}
 				}
 				if (this.errorHandler == null) {
 					throw e;

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

@@ -86,6 +86,7 @@
  * @author Martin Dam
  * @author Artem Bilan
  * @author Loic Talhouarne
+ * @author Tom van den Berge
  */
 public class KafkaMessageListenerContainerTests {
 
@@ -121,9 +122,13 @@ public class KafkaMessageListenerContainerTests {
 
 	private static String topic17 = "testTopic17";
 
+	private static String topic18 = "testTopic18";
+
+	
 	@ClassRule
 	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic3, topic4, topic5,
-			topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16, topic17);
+			topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16, topic17,
+			topic18);
 
 	@Rule
 	public TestName testName = new TestName();
@@ -1435,6 +1440,101 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 		logger.info("Stop manual ack rebalance");
 	}
 
+	@Test
+	public void testRebalanceAfterFailedRecord() throws Exception {
+		logger.info("Start rebalance after failed record");
+		Map<String, Object> props = KafkaTestUtils.consumerProps("test18", "false", embeddedKafka);
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
+		ContainerProperties containerProps = new ContainerProperties(topic18);
+		final List<AtomicInteger> counts = new ArrayList<>();
+		counts.add(new AtomicInteger());
+		counts.add(new AtomicInteger());
+		containerProps.setMessageListener(new MessageListener<Integer, String>() {
+
+			@Override
+			public void onMessage(ConsumerRecord<Integer, String> message) {
+				// The 1st message per partition fails
+				if (counts.get(message.partition()).incrementAndGet() < 2) {
+					throw new RuntimeException("Failure wile processing message");
+				}
+			}
+		});
+		containerProps.setSyncCommits(true);
+		containerProps.setAckMode(AckMode.RECORD);
+		final CountDownLatch rebalanceLatch = new CountDownLatch(2);
+		containerProps.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
+
+			@Override
+			public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+			}
+
+			@Override
+			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+				logger.info("manual ack: assigned " + partitions);
+				rebalanceLatch.countDown();
+			}
+		});
+
+		CountDownLatch stubbingComplete1 = new CountDownLatch(1);
+		KafkaMessageListenerContainer<Integer, String> container1 =
+				spyOnContainer(new KafkaMessageListenerContainer<>(cf, containerProps), stubbingComplete1);
+		container1.setBeanName("testRebalanceAfterFailedRecord");
+		container1.start();
+		Consumer<?, ?> containerConsumer = spyOnConsumer(container1);
+		final CountDownLatch commitLatch = new CountDownLatch(2);
+		willAnswer(invocation -> {
+
+			@SuppressWarnings({ "unchecked" })
+			Map<TopicPartition, OffsetAndMetadata> map = invocation.getArgument(0);
+			try {
+				return invocation.callRealMethod();
+			}
+			finally {
+				for (Entry<TopicPartition, OffsetAndMetadata> entry : map.entrySet()) {
+					// Decrement when the last (successful) has been committed
+					if (entry.getValue().offset() == 2) {
+						commitLatch.countDown();
+					}
+				}
+			}
+
+		}).given(containerConsumer).commitSync(any());
+		stubbingComplete1.countDown();
+		ContainerTestUtils.waitForAssignment(container1, embeddedKafka.getPartitionsPerTopic());
+
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(topic18);
+		template.sendDefault(0, 0, "foo");
+		template.sendDefault(1, 0, "baz");
+		template.sendDefault(0, 0, "bar");
+		template.sendDefault(1, 0, "qux");
+		template.flush();
+
+		// Wait until both partitions have committed offset 2 (i.e. the last message)
+		assertThat(commitLatch.await(30, TimeUnit.SECONDS)).isTrue();
+
+		// Start a 2nd consumer, triggering a rebalance
+		KafkaMessageListenerContainer<Integer, String> container2 =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container2.setBeanName("testRebalanceAfterFailedRecord2");
+		container2.start();
+		// Wait until both consumers have finished rebalancing
+		assertThat(rebalanceLatch.await(60, TimeUnit.SECONDS)).isTrue();
+
+		// Stop both consumers
+		container1.stop();
+		container2.stop();
+		Consumer<Integer, String> consumer = cf.createConsumer();
+		consumer.assign(Arrays.asList(new TopicPartition(topic18, 0), new TopicPartition(topic18, 1)));
+
+		// Verify that offset of both partitions is the highest committed offset
+		assertThat(consumer.position(new TopicPartition(topic18, 0))).isEqualTo(2);
+		assertThat(consumer.position(new TopicPartition(topic18, 1))).isEqualTo(2);
+		consumer.close();
+		logger.info("Stop rebalance after failed record");
+	}
 	private Consumer<?, ?> spyOnConsumer(KafkaMessageListenerContainer<Integer, String> container) {
 		Consumer<?, ?> consumer = spy(
 				KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class));
@@ -1443,8 +1543,10 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 		return consumer;
 	}
 
-	private KafkaMessageListenerContainer<Integer, String> spyOnContainer(KafkaMessageListenerContainer<Integer, String> container,
+	private KafkaMessageListenerContainer<Integer, String> spyOnContainer(
+			KafkaMessageListenerContainer<Integer, String> container,
 			final CountDownLatch stubbingComplete) {
+
 		KafkaMessageListenerContainer<Integer, String> spy = spy(container);
 		willAnswer(i -> {
 			if (stubbingComplete.getCount() > 0 && Thread.currentThread().getName().endsWith("-C-1")) {