GH-990: Run AfterRollbackProcessor in Tx

Resolves https://github.com/spring-projects/spring-kafka/issues/990

Provide a mechanism to start a new transaction within which to invoke
the processor, so if it recovers the failed record, its offset can
be sent to the transaction.

**cherry-pick to 2.2.x**

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

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -44,11 +44,17 @@ public interface AfterRollbackProcessor<K, V> {
 	 * processing individual records; this allows the processor to recover (skip) the
 	 * failed record rather than re-seeking it. This is not possible with a batch listener
 	 * since only the listener itself knows which record in the batch keeps failing.
+	 * IMPORTANT: If invoked in a transaction when the listener was invoked with a single
+	 * record, the transaction id will be based on the container group.id and the
+	 * topic/partition of the failed record, to avoid issues with zombie fencing. So,
+	 * generally, only its offset should be sent to the transaction. For other behavior
+	 * the process method should manage its own transaction.
 	 * @param records the records.
 	 * @param consumer the consumer.
 	 * @param exception the exception
 	 * @param recoverable the recoverable.
 	 * @since 2.2
+	 * @see #isProcessInTransaction()
 	 */
 	void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer, Exception exception, boolean recoverable);
 
@@ -61,4 +67,17 @@ public interface AfterRollbackProcessor<K, V> {
 		// NOSONAR
 	}
 
+	/**
+	 * Return true to invoke {@link #process(List, Consumer, Exception, boolean)} in a new
+	 * transaction. Because the container cannot infer the desired behavior, the processor
+	 * is responsible for sending the offset to the transaction if it decides to skip the
+	 * failing record.
+	 * @return true to run in a transaction; default false.
+	 * @since 2.2.5
+	 * @see #process(List, Consumer, Exception, boolean)
+	 */
+	default boolean isProcessInTransaction() {
+		return false;
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -120,9 +120,15 @@ public class DeadLetterPublishingRecoverer implements BiConsumer<ConsumerRecord<
 				record.key(), record.value(), headers);
 	}
 
-	private void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> template) {
+	/**
+	 * Override this if you want more than just logging of the send result.
+	 * @param outRecord the record to send.
+	 * @param kafkaTemplate the template.
+	 * @since 2.2.5
+	 */
+	protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate) {
 		try {
-			template.send(outRecord).addCallback(result -> {
+			kafkaTemplate.send(outRecord).addCallback(result -> {
 				if (logger.isDebugEnabled()) {
 					logger.debug("Successful dead-letter publication: " + result);
 				}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.listener;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.function.BiConsumer;
 
@@ -23,7 +24,10 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SeekUtils;
 import org.springframework.lang.Nullable;
 
@@ -49,6 +53,10 @@ public class DefaultAfterRollbackProcessor<K, V> implements AfterRollbackProcess
 
 	private final FailedRecordTracker failureTracker;
 
+	private boolean processInTransaction;
+
+	private KafkaTemplate<K, V> kafkaTemplate;
+
 	/**
 	 * Construct an instance with the default recoverer which simply logs the record after
 	 * {@value SeekUtils#DEFAULT_MAX_FAILURES} (maxFailures) have occurred for a
@@ -96,8 +104,42 @@ public class DefaultAfterRollbackProcessor<K, V> implements AfterRollbackProcess
 	@Override
 	public void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer, Exception exception,
 			boolean recoverable) {
-		SeekUtils.doSeeks(((List) records),
-				consumer, exception, recoverable, this.failureTracker::skip, logger);
+
+		if (SeekUtils.doSeeks(((List) records), consumer, exception, recoverable, this.failureTracker::skip, logger)
+				&& this.kafkaTemplate != null && this.kafkaTemplate.isTransactional()) {
+			ConsumerRecord<K, V> skipped = records.get(0);
+			this.kafkaTemplate.sendOffsetsToTransaction(
+					Collections.singletonMap(new TopicPartition(skipped.topic(), skipped.partition()),
+							new OffsetAndMetadata(skipped.offset() + 1)));
+		}
+	}
+
+	@Override
+	public boolean isProcessInTransaction() {
+		return this.processInTransaction;
+	}
+
+	/**
+	 * Set to true to run the {@link #process(List, Consumer, Exception, boolean)}
+	 * method in a transaction. Requires a {@link KafkaTemplate}.
+	 * @param processInTransaction true to process in a transaction.
+	 * @since 2.2.5
+	 * @see #process(List, Consumer, Exception, boolean)
+	 * @see #setKafkaTemplate(KafkaTemplate)
+	 */
+	public void setProcessInTransaction(boolean processInTransaction) {
+		this.processInTransaction = processInTransaction;
+	}
+
+	/**
+	 * Set a {@link KafkaTemplate} to use to send the offset of a recovered record
+	 * to a transaction.
+	 * @param kafkaTemplate the template
+	 * @since 2.2.5
+	 * @see #setProcessInTransaction(boolean)
+	 */
+	public void setKafkaTemplate(KafkaTemplate<K, V> kafkaTemplate) {
+		this.kafkaTemplate = kafkaTemplate;
 	}
 
 	@Override

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

@@ -573,14 +573,14 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 			}
 		}
 
-		private void subscribeOrAssignTopics(final Consumer<? super K, ? super V> consumer) {
+		private void subscribeOrAssignTopics(final Consumer<? super K, ? super V> subscribingConsumer) {
 			if (KafkaMessageListenerContainer.this.topicPartitions == null) {
 				ConsumerRebalanceListener rebalanceListener = new ListenerConsumerRebalanceListener();
 				if (this.containerProperties.getTopicPattern() != null) {
-					consumer.subscribe(this.containerProperties.getTopicPattern(), rebalanceListener);
+					subscribingConsumer.subscribe(this.containerProperties.getTopicPattern(), rebalanceListener);
 				}
 				else {
-					consumer.subscribe(Arrays.asList(this.containerProperties.getTopics()), rebalanceListener);
+					subscribingConsumer.subscribe(Arrays.asList(this.containerProperties.getTopics()), rebalanceListener);
 				}
 			}
 			else {
@@ -592,7 +592,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 							new OffsetMetadata(topicPartition.initialOffset(), topicPartition.isRelativeToCurrent(),
 									topicPartition.getPosition()));
 				}
-				consumer.assign(new ArrayList<>(this.definedPartitions.keySet()));
+				subscribingConsumer.assign(new ArrayList<>(this.definedPartitions.keySet()));
 			}
 		}
 
@@ -695,7 +695,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 				try {
 					pollAndInvoke();
 				}
-				catch (WakeupException e) {
+				catch (@SuppressWarnings("unused") WakeupException e) {
 					// Ignore, we're stopping
 				}
 				catch (NoOffsetForPartitionException nofpe) {
@@ -814,7 +814,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 					try {
 						this.consumer.unsubscribe();
 					}
-					catch (WakeupException e) {
+					catch (@SuppressWarnings("unused") WakeupException e) {
 						// No-op. Continue process
 					}
 				}
@@ -900,7 +900,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 					try {
 						ackImmediate(record);
 					}
-					catch (WakeupException e) {
+					catch (@SuppressWarnings("unused") WakeupException e) {
 						// ignore - not polling
 					}
 				}
@@ -950,6 +950,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 		@SuppressWarnings({ UNCHECKED, RAW_TYPES })
 		private void invokeBatchListenerInTx(final ConsumerRecords<K, V> records,
 				final List<ConsumerRecord<K, V>> recordList) {
+
 			try {
 				this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
 
@@ -972,15 +973,34 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 				this.logger.error("Transaction rolled back", e);
 				AfterRollbackProcessor<K, V> afterRollbackProcessorToUse =
 						(AfterRollbackProcessor<K, V>) getAfterRollbackProcessor();
-				if (recordList == null) {
-					afterRollbackProcessorToUse.process(createRecordList(records), this.consumer, e, false);
+				if (afterRollbackProcessorToUse.isProcessInTransaction() && this.transactionTemplate != null) {
+					this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
+
+						@Override
+						protected void doInTransactionWithoutResult(TransactionStatus status) {
+							batchAfterRollback(records, recordList, e, afterRollbackProcessorToUse);
+						}
+
+					});
 				}
 				else {
-					afterRollbackProcessorToUse.process(recordList, this.consumer, e, false);
+					batchAfterRollback(records, recordList, e, afterRollbackProcessorToUse);
 				}
 			}
 		}
 
+		private void batchAfterRollback(final ConsumerRecords<K, V> records,
+				final List<ConsumerRecord<K, V>> recordList, RuntimeException e,
+				AfterRollbackProcessor<K, V> afterRollbackProcessorToUse) {
+
+			if (recordList == null) {
+				afterRollbackProcessorToUse.process(createRecordList(records), this.consumer, e, false);
+			}
+			else {
+				afterRollbackProcessorToUse.process(recordList, this.consumer, e, false);
+			}
+		}
+
 		private List<ConsumerRecord<K, V>> createRecordList(final ConsumerRecords<K, V> records) {
 			return StreamSupport.stream(records.spliterator(), false)
 					.collect(Collectors.toList());
@@ -1020,7 +1040,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 					throw er;
 				}
 			}
-			catch (InterruptedException e) {
+			catch (@SuppressWarnings("unused") InterruptedException e) {
 				Thread.currentThread().interrupt();
 			}
 			return null;
@@ -1101,7 +1121,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 		 * Invoke the listener with each record in a separate transaction.
 		 * @param records the records.
 		 */
-		@SuppressWarnings({ UNCHECKED, RAW_TYPES })
+		@SuppressWarnings(RAW_TYPES)
 		private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 			while (iterator.hasNext()) {
@@ -1132,13 +1152,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 				}
 				catch (RuntimeException e) {
 					this.logger.error("Transaction rolled back", e);
-					List<ConsumerRecord<K, V>> unprocessed = new ArrayList<>();
-					unprocessed.add(record);
-					while (iterator.hasNext()) {
-						unprocessed.add(iterator.next());
-					}
-					((AfterRollbackProcessor<K, V>) getAfterRollbackProcessor())
-							.process(unprocessed, this.consumer, e, true);
+					recordAfterRollback(iterator, record, e);
 				}
 				finally {
 					TransactionSupport.clearTransactionIdSuffix();
@@ -1146,6 +1160,32 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 			}
 		}
 
+		private void recordAfterRollback(Iterator<ConsumerRecord<K, V>> iterator, final ConsumerRecord<K, V> record,
+				RuntimeException e) {
+
+			List<ConsumerRecord<K, V>> unprocessed = new ArrayList<>();
+			unprocessed.add(record);
+			while (iterator.hasNext()) {
+				unprocessed.add(iterator.next());
+			}
+			@SuppressWarnings(UNCHECKED)
+			AfterRollbackProcessor<K, V> afterRollbackProcessorToUse =
+					(AfterRollbackProcessor<K, V>) getAfterRollbackProcessor();
+			if (afterRollbackProcessorToUse.isProcessInTransaction() && this.transactionTemplate != null) {
+				this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
+
+					@Override
+					protected void doInTransactionWithoutResult(TransactionStatus status) {
+						afterRollbackProcessorToUse.process(unprocessed, ListenerConsumer.this.consumer, e, true);
+					}
+
+				});
+			}
+			else {
+				afterRollbackProcessorToUse.process(unprocessed, this.consumer, e, true);
+			}
+		}
+
 		private void doInvokeWithRecords(final ConsumerRecords<K, V> records) {
 			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 			while (iterator.hasNext()) {
@@ -1499,7 +1539,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 						this.consumer.commitAsync(commits, this.commitCallback);
 					}
 				}
-				catch (WakeupException e) {
+				catch (@SuppressWarnings("unused") WakeupException e) {
 					// ignore - not polling
 					this.logger.debug("Woken up during commit");
 				}

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

@@ -18,6 +18,7 @@ package org.springframework.kafka.listener;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -494,6 +495,7 @@ public class TransactionalContainerTests {
 		consumer.close();
 	}
 
+	@SuppressWarnings("unchecked")
 	@Test
 	public void testMaxFailures() throws Exception {
 		logger.info("Start testMaxFailures");
@@ -520,14 +522,15 @@ public class TransactionalContainerTests {
 			latch.countDown();
 		});
 
-		@SuppressWarnings({ "rawtypes", "unchecked" })
+		@SuppressWarnings({ "rawtypes" })
 		KafkaTransactionManager tm = new KafkaTransactionManager(pf);
 		containerProps.setTransactionManager(tm);
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.setBeanName("testMaxFailures");
 		final CountDownLatch recoverLatch = new CountDownLatch(1);
-		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template) {
+		final KafkaTemplate<Object, Object> dlTemplate = spy(new KafkaTemplate<>(pf));
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(dlTemplate) {
 
 			@Override
 			public void accept(ConsumerRecord<?, ?> record, Exception exception) {
@@ -536,8 +539,10 @@ public class TransactionalContainerTests {
 			}
 
 		};
-		DefaultAfterRollbackProcessor<Integer, String> afterRollbackProcessor =
+		DefaultAfterRollbackProcessor<Object, Object> afterRollbackProcessor =
 				spy(new DefaultAfterRollbackProcessor<>(recoverer, 3));
+		afterRollbackProcessor.setProcessInTransaction(true);
+		afterRollbackProcessor.setKafkaTemplate(dlTemplate);
 		container.setAfterRollbackProcessor(afterRollbackProcessor);
 		final CountDownLatch stopLatch = new CountDownLatch(1);
 		container.setApplicationEventPublisher(e -> {
@@ -579,7 +584,12 @@ public class TransactionalContainerTests {
 		assertThat(headers.get("baz")).isEqualTo("qux".getBytes());
 		pf.destroy();
 		assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		verify(afterRollbackProcessor, times(3)).isProcessInTransaction();
+		verify(afterRollbackProcessor, times(3)).process(any(), any(), any(), anyBoolean());
 		verify(afterRollbackProcessor).clearThreadState();
+		verify(dlTemplate).send(any(ProducerRecord.class));
+		verify(dlTemplate).sendOffsetsToTransaction(
+				Collections.singletonMap(new TopicPartition(topic3, 0), new OffsetAndMetadata(1L)));
 		logger.info("Stop testMaxAttempts");
 	}
 

src/reference/asciidoc/kafka.adoc

@@ -2834,6 +2834,10 @@ In such cases, the application listener must handle a record that keeps failing.
 
 See also <<dead-letters>>.
 
+Starting with version 2.2.5, the `DefaultAfterRollbackProcessor` can be invoked in a new transaction (started after the failed transaction rolls back).
+Then, if you are using the `DeadLetterPublishingRecoverer` to publish a failed record, the processor will send the recovered record's offset in the original topic/partition to the transaction.
+To enable this feature, set the `processInTransaction` and `kafkaTemplate` properties on the `DefaultAfterRollbackProcessor`.
+
 [[dead-letters]]
 ===== Publishing Dead-letter Records