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

Author: garyrussell

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

@@ -1112,12 +1112,7 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> recor
 						this.errorHandler.handle(e, record, this.consumer);
 					}
 					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

@@ -326,6 +326,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

@@ -99,16 +99,21 @@ public class TransactionalContainerTests {
 
 	@Test
 	public void testConsumeAndProduceTransactionKTM() throws Exception {
-		testConsumeAndProduceTransactionGuts(false);
+		testConsumeAndProduceTransactionGuts(false, false);
 	}
 
 	@Test
 	public void testConsumeAndProduceTransactionKCTM() throws Exception {
-		testConsumeAndProduceTransactionGuts(true);
+		testConsumeAndProduceTransactionGuts(true, false);
+	}
+
+	@Test
+	public void testConsumeAndProduceTransactionHandleError() throws Exception {
+		testConsumeAndProduceTransactionGuts(false, true);
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
-	private void testConsumeAndProduceTransactionGuts(boolean chained) 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 -> {
@@ -150,7 +155,13 @@ private void testConsumeAndProduceTransactionGuts(boolean chained) throws Except
 		final KafkaTemplate template = new KafkaTemplate(pf);
 		props.setMessageListener((MessageListener) m -> {
 			template.send("bar", "baz");
+			if (handleError) {
+				throw new RuntimeException("fail");
+			}
 		});
+		if (handleError) {
+			props.setErrorHandler((e, data) -> { });
+		}
 		KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, props);
 		container.setBeanName("commit");
 		container.start();