GH-683: Fix Transactions with ErrorHandler

Fixes #683

If the error handler handles the error, the offset should be sent to the tx.

**cherry-pick to 2.0.x, 1.3.x**

* Polish ackOnError Javadoc to reflect transactions

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

Author: garyrussell

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

@@ -966,12 +966,7 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> recor
 				try {
 					this.errorHandler.handle(e, record);
 					if (producer != null) {
-						try {
-							sendOffsetsToTransaction(producer);
-						}
-						catch (Exception e1) {
-							this.logger.error("Send offsets to transaction failed", e1);
-						}
+						ackCurrent(record, producer);
 					}
 				}
 				catch (RuntimeException ee) {

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

@@ -317,6 +317,13 @@ public void setIdleEventInterval(Long idleEventInterval) {
 	 * offset of the failed message anyway, so this option has limited applicability.
 	 * Perhaps useful for a component that starts throwing exceptions consistently;
 	 * allowing it to resume when restarted from the last successfully processed message.
+	 * <p>
+	 * Does not apply when transactions are used - in that case, whether or not the
+	 * offsets are sent to the transaction depends on whether the transaction is committed
+	 * or rolled back. If a listener throws an exception, the transaction will normally
+	 * be rolled back unless an error handler is provided that handles the error and
+	 * exits normally; in which case the offsets are sent to the transaction and the
+	 * transaction is committed.
 	 * @param ackOnError whether the container should acknowledge messages that throw
 	 * exceptions.
 	 */

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -63,7 +63,6 @@
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
-import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
 import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
@@ -93,17 +92,22 @@ public class TransactionalContainerTests {
 	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(3, true, topic1, topic2);
 
 	@Test
-	public void testConsumeAndProduceTransactionBatch() throws Exception {
-		testConsumeAndProduceTransactionGuts(AckMode.BATCH);
+	public void testConsumeAndProduceTransactionKTM() throws Exception {
+		testConsumeAndProduceTransactionGuts(false, false);
 	}
 
 	@Test
-	public void testConsumeAndProduceTransactionRecord() throws Exception {
-		testConsumeAndProduceTransactionGuts(AckMode.RECORD);
+	public void testConsumeAndProduceTransactionKCTM() throws Exception {
+		testConsumeAndProduceTransactionGuts(true, false);
+	}
+
+	@Test
+	public void testConsumeAndProduceTransactionHandleError() throws Exception {
+		testConsumeAndProduceTransactionGuts(false, true);
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
-	private void testConsumeAndProduceTransactionGuts(AckMode ackMode) throws Exception {
+	private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handleError) throws Exception {
 		Consumer consumer = mock(Consumer.class);
 		final TopicPartition topicPartition = new TopicPartition("foo", 0);
 		willAnswer(i -> {
@@ -141,8 +145,13 @@ private void testConsumeAndProduceTransactionGuts(AckMode ackMode) throws Except
 		final KafkaTemplate template = new KafkaTemplate(pf);
 		props.setMessageListener((MessageListener) m -> {
 			template.send("bar", "baz");
+			if (handleError) {
+				throw new RuntimeException("fail");
+			}
 		});
-		props.setAckMode(ackMode);
+		if (handleError) {
+			props.setErrorHandler((e, data) -> { });
+		}
 		KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, props);
 		container.setBeanName("commit");
 		container.start();