GH-87: Add ackOnError and fix ack commit issues

Fixes #87 (https://github.com/spring-projects/spring-kafka/issues/87)

- allow for acking on both success and error (make the latter conditional upon `ackOnError`)
- fixes an issue where `processCommits()` wasn't handled on invoker stop, allowing for missed commits

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

@@ -470,6 +470,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			}
 			// handle the last manual acks, after the listeners have closed
 			handleManualAcks();
+			processCommits();
 			if (this.offsets.size() > 0) {
 				// we always commit after stopping the invoker
 				commitIfNecessary();
@@ -602,6 +603,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 					}
 				}
 				catch (Exception e) {
+					if (this.containerProperties.isAckOnError()) {
+						this.acks.add(record);
+					}
 					if (this.containerProperties.getErrorHandler() != null) {
 						this.containerProperties.getErrorHandler().handle(e, record);
 					}

spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java

@@ -181,6 +181,8 @@ public class ContainerProperties {
 	 */
 	private boolean syncCommits = true;
 
+	private boolean ackOnError = true;
+
 	private Long idleEventInterval;
 
 	public ContainerProperties(String... topics) {
@@ -385,6 +387,21 @@ public class ContainerProperties {
 		this.idleEventInterval = idleEventInterval;
 	}
 
+	/**
+	 * Set whether the container should ack messages that throw exceptions or not. This
+	 * works in conjunction with {@link #ackMode} and is effective only when auto ack is
+	 * false; it is not applicable to manual acks. When this property is set to
+	 * {@code true}, all messages handled will be acked. When set to {@code false}, acks
+	 * will be produced only for successful messages. This allows a component that starts
+	 * throwing exceptions consistently to resume from the last successfully processed
+	 * message. Manual acks will be always be applied.
+	 * @param ackOnError whether the container should acknowledge messages that throw
+	 * exceptions.
+	 */
+	public void setAckOnError(boolean ackOnError) {
+		this.ackOnError = ackOnError;
+	}
+
 	public String[] getTopics() {
 		return this.topics;
 	}
@@ -473,4 +490,7 @@ public class ContainerProperties {
 		return this.idleEventInterval;
 	}
 
+	public boolean isAckOnError() {
+		return this.ackOnError;
+	}
 }

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

@@ -22,6 +22,7 @@ import static org.mockito.Matchers.anyLong;
 import static org.mockito.Mockito.mock;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.BitSet;
 import java.util.Collection;
 import java.util.List;
@@ -65,6 +66,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils;
  * @author Gary Russell
  * @author Artem Bilan
  * @author Jerome Mirc
+ * @author Marius Bogoevici
  *
  */
 public class ConcurrentMessageListenerContainerTests {
@@ -87,9 +89,11 @@ public class ConcurrentMessageListenerContainerTests {
 
 	private static String topic8 = "testTopic8";
 
+	private static String topic9 = "testTopic9";
+
 	@ClassRule
 	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5,
-			topic6, topic7, topic8);
+			topic6, topic7, topic8, topic9);
 
 	@Test
 	public void testAutoCommit() throws Exception {
@@ -608,4 +612,56 @@ public class ConcurrentMessageListenerContainerTests {
 
 	}
 
+
+	@Test
+	public void testAckOnErrorRecord() throws Exception {
+		logger.info("Start ack on error");
+		Map<String, Object> props = KafkaTestUtils.consumerProps("test9", "false", embeddedKafka);
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
+		final CountDownLatch latch = new CountDownLatch(4);
+		ContainerProperties containerProps = new ContainerProperties(topic9);
+		containerProps.setMessageListener(new MessageListener<Integer, String>() {
+
+			@Override
+			public void onMessage(ConsumerRecord<Integer, String> message) {
+				logger.info("auto ack on error: " + message);
+				latch.countDown();
+				if (message.value().startsWith("b")) {
+					throw new RuntimeException();
+				}
+			}
+		});
+		containerProps.setSyncCommits(true);
+		containerProps.setAckMode(AckMode.RECORD);
+		containerProps.setAckOnError(false);
+		ConcurrentMessageListenerContainer<Integer, String> container = new ConcurrentMessageListenerContainer<>(cf,
+				containerProps);
+		container.setConcurrency(2);
+		container.setBeanName("testAckOnError");
+		container.start();
+		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(topic9);
+		template.sendDefault(0, 0, "foo");
+		template.sendDefault(1, 0, "bar");
+		template.sendDefault(0, 0, "baz");
+		template.sendDefault(1, 0, "qux");
+		template.flush();
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+		Consumer<Integer, String> consumer = cf.createConsumer();
+		consumer.assign(Arrays.asList(new TopicPartition(topic9, 0), new TopicPartition(topic9, 1)));
+		// this consumer is positioned at 1, the next offset after the successfully
+		// processed 'foo'
+		// it has not been updated because 'bar' failed
+		assertThat(consumer.position(new TopicPartition(topic9, 0))).isEqualTo(1);
+		// this consumer is positioned at 1, the next offset after the successfully
+		// processed 'qux'
+		// it has been updated even 'baz' failed
+		assertThat(consumer.position(new TopicPartition(topic9, 1))).isEqualTo(2);
+		logger.info("Stop ack on error");
+	}
+
 }