Try/Catch around error recovery

- if a recoverer `BiConsumer` threw an exception, the seeks would not be performed
- also add try/catch around after rollback processor calls

** Manual backport PR required due to new BackOff logic on master **

Author: garyrussell

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

@@ -43,24 +43,31 @@ class FailedRecordTracker {
 
 	private final BackOff backOff;
 
+	private final LogAccessor logger;
+
 	FailedRecordTracker(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff,
 			LogAccessor logger) {
 
 		Assert.notNull(backOff, "'backOff' cannot be null");
 		if (recoverer == null) {
-			this.recoverer = (rec, thr) -> logger.error(thr, "Backoff " + this.failures.get().getBackOffExecution()
+			FailedRecord failedRecord = this.failures.get();
+			this.recoverer = (rec, thr) -> logger.error(thr, "Backoff "
+				+ (failedRecord == null
+					? "none"
+					: failedRecord.getBackOffExecution())
 				+ " exhausted for " + rec);
 		}
 		else {
 			this.recoverer = recoverer;
 		}
 		this.noRetries = backOff.start().nextBackOff() == BackOffExecution.STOP;
 		this.backOff = backOff;
+		this.logger = logger;
 	}
 
 	boolean skip(ConsumerRecord<?, ?> record, Exception exception) {
 		if (this.noRetries) {
-			this.recoverer.accept(record, exception);
+			recover(record, exception);
 			return true;
 		}
 		FailedRecord failedRecord = this.failures.get();
@@ -79,12 +86,21 @@ boolean skip(ConsumerRecord<?, ?> record, Exception exception) {
 			return false;
 		}
 		else {
-			this.recoverer.accept(record, exception);
+			recover(record, exception);
 			this.failures.remove();
 			return true;
 		}
 	}
 
+	private void recover(ConsumerRecord<?, ?> record, Exception exception) {
+		try {
+			this.recoverer.accept(record, exception);
+		}
+		catch (Exception ex) {
+			this.logger.error(ex, "Recoverer threw exception");
+		}
+	}
+
 	private boolean newFailure(ConsumerRecord<?, ?> record, FailedRecord failedRecord) {
 		return !failedRecord.getTopic().equals(record.topic())
 				|| failedRecord.getPartition() != record.partition()

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

@@ -1133,11 +1133,16 @@ 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);
+			try {
+				if (recordList == null) {
+					afterRollbackProcessorToUse.process(createRecordList(records), this.consumer, e, false);
+				}
+				else {
+					afterRollbackProcessorToUse.process(recordList, this.consumer, e, false);
+				}
 			}
-			else {
-				afterRollbackProcessorToUse.process(recordList, this.consumer, e, false);
+			catch (Exception ex) {
+				this.logger.error(ex, "AfterRollbackProcessor threw exception");
 			}
 		}
 
@@ -1321,7 +1326,12 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
 				});
 			}
 			else {
-				afterRollbackProcessorToUse.process(unprocessed, this.consumer, e, true);
+				try {
+					afterRollbackProcessorToUse.process(unprocessed, this.consumer, e, true);
+				}
+				catch (Exception ex) {
+					this.logger.error(ex, "AfterRollbackProcessor threw exception");
+				}
 			}
 		}
 

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

@@ -46,6 +46,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
 
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -112,9 +113,11 @@ public class TransactionalContainerTests {
 
 	private static String topic3DLT = "txTopic3.DLT";
 
+	private static String topic4 = "txTopic4";
+
 	@ClassRule
 	public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, topic1, topic2, topic3,
-				topic3DLT)
+				topic3DLT, topic4)
 			.brokerProperty(KafkaConfig.TransactionsTopicReplicationFactorProp(), "1")
 			.brokerProperty(KafkaConfig.TransactionsTopicMinISRProp(), "1");
 
@@ -595,6 +598,69 @@ public void accept(ConsumerRecord<?, ?> record, Exception exception) {
 		logger.info("Stop testMaxAttempts");
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testRollbackProcessorCrash() throws Exception {
+		logger.info("Start testRollbackNoRetries");
+		Map<String, Object> props = KafkaTestUtils.consumerProps("testRollbackNoRetries", "false", embeddedKafka);
+		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
+		props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
+		ContainerProperties containerProps = new ContainerProperties(topic4);
+		containerProps.setPollTimeout(10_000);
+
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		senderProps.put(ProducerConfig.RETRIES_CONFIG, 1);
+		DefaultKafkaProducerFactory<Object, Object> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		pf.setTransactionIdPrefix("noRetries.");
+		final KafkaTemplate<Object, Object> template = new KafkaTemplate<>(pf);
+		final CountDownLatch latch = new CountDownLatch(1);
+		AtomicReference<String> data = new AtomicReference<>();
+		containerProps.setMessageListener((MessageListener<Integer, String>) message -> {
+			data.set(message.value());
+			if (message.offset() == 0) {
+				throw new RuntimeException("fail for no retry");
+			}
+			latch.countDown();
+		});
+
+		@SuppressWarnings({ "rawtypes" })
+		KafkaTransactionManager tm = new KafkaTransactionManager(pf);
+		containerProps.setTransactionManager(tm);
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.setBeanName("testRollbackNoRetries");
+		BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer = (rec, ex) -> {
+			throw new RuntimeException("arbp fail");
+		};
+		DefaultAfterRollbackProcessor<Object, Object> afterRollbackProcessor =
+				spy(new DefaultAfterRollbackProcessor<>(recoverer, new FixedBackOff(0L, 0L)));
+		container.setAfterRollbackProcessor(afterRollbackProcessor);
+		final CountDownLatch stopLatch = new CountDownLatch(1);
+		container.setApplicationEventPublisher(e -> {
+			if (e instanceof ConsumerStoppedEvent) {
+				stopLatch.countDown();
+			}
+		});
+		container.start();
+
+		template.setDefaultTopic(topic4);
+		template.executeInTransaction(t -> {
+			RecordHeaders headers = new RecordHeaders(new RecordHeader[] { new RecordHeader("baz", "qux".getBytes()) });
+			ProducerRecord<Object, Object> record = new ProducerRecord<>(topic4, 0, 0, "foo", headers);
+			template.send(record);
+			template.sendDefault(0, 0, "bar");
+			return null;
+		});
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		assertThat(data.get()).isEqualTo("bar");
+		container.stop();
+		pf.destroy();
+		assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		logger.info("Stop testRollbackNoRetries");
+	}
+
 	@SuppressWarnings("serial")
 	public static class SomeOtherTransactionManager extends AbstractPlatformTransactionManager {