GH-1259: Handle Failed Record Recovery

Resolves #1259

Previously if the recoverer in a `SeekToCurrentErrorHandler` or
`DefaultAfterRollbackProcessor` failed to recover a record, the
record could be lost; the `FailedRecordTracker` simply logged
the exception.

Change the `SeekUtils` to detect a failure in the recoverer (actually
any failure when determining if the failed record should be recovered)
and include the failed record in the seeks.

In this way the recovery will be attempted once more on each delivery
attempt.

**cherry-pick to 2.2.x**

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordTracker.java
#	spring-kafka/src/main/java/org/springframework/kafka/support/SeekUtils.java
#	src/reference/asciidoc/kafka.adoc

Author: garyrussell

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

@@ -60,7 +60,7 @@ class FailedRecordTracker {
 
 	boolean skip(ConsumerRecord<?, ?> record, Exception exception) {
 		if (this.noRetries) {
-			recover(record, exception);
+			this.recoverer.accept(record, exception);
 			return true;
 		}
 		Map<TopicPartition, FailedRecord> map = this.failures.get();
@@ -76,7 +76,7 @@ boolean skip(ConsumerRecord<?, ?> record, Exception exception) {
 			return false;
 		}
 		else if (this.maxFailures > 0 && failedRecord.incrementAndGet() >= this.maxFailures) {
-			recover(record, exception);
+			this.recoverer.accept(record, exception);
 			map.remove(topicPartition);
 			if (map.isEmpty()) {
 				this.failures.remove();
@@ -88,15 +88,6 @@ else if (this.maxFailures > 0 && failedRecord.incrementAndGet() >= this.maxFailu
 		}
 	}
 
-	private void recover(ConsumerRecord<?, ?> record, Exception exception) {
-		try {
-			this.recoverer.accept(record, exception);
-		}
-		catch (Exception ex) {
-			this.logger.error("Recoverer threw exception", ex);
-		}
-	}
-
 	void clearThreadState() {
 		this.failures.remove();
 	}

spring-kafka/src/main/java/org/springframework/kafka/support/SeekUtils.java

@@ -63,7 +63,14 @@ public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?>
 		AtomicBoolean skipped = new AtomicBoolean();
 		records.forEach(record ->  {
 			if (recoverable && first.get()) {
-				skipped.set(skipper.test(record, exception));
+				try {
+					boolean test = skipper.test(record, exception);
+					skipped.set(test);
+				}
+				catch (Exception ex) {
+					logger.error("Failed to determine if this record should be recovererd, including in seeks", ex);
+					skipped.set(false);
+				}
 				if (skipped.get() && logger.isDebugEnabled()) {
 					logger.debug("Skipping seek of: " + record);
 				}

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

@@ -533,10 +533,14 @@ public void testMaxFailures() throws Exception {
 		container.setBeanName("testMaxFailures");
 		final CountDownLatch recoverLatch = new CountDownLatch(1);
 		final KafkaTemplate<Object, Object> dlTemplate = spy(new KafkaTemplate<>(pf));
+		AtomicBoolean recovererShouldFail = new AtomicBoolean(true);
 		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(dlTemplate) {
 
 			@Override
 			public void accept(ConsumerRecord<?, ?> record, Exception exception) {
+				if (recovererShouldFail.getAndSet(false)) {
+					throw new RuntimeException("test recoverer failure");
+				}
 				super.accept(record, exception);
 				recoverLatch.countDown();
 			}
@@ -587,8 +591,8 @@ public void accept(ConsumerRecord<?, ?> record, Exception exception) {
 		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, times(4)).isProcessInTransaction();
+		verify(afterRollbackProcessor, times(4)).process(any(), any(), any(), anyBoolean());
 		verify(afterRollbackProcessor).clearThreadState();
 		verify(dlTemplate).send(any(ProducerRecord.class));
 		verify(dlTemplate).sendOffsetsToTransaction(
@@ -629,8 +633,11 @@ public void testRollbackProcessorCrash() throws Exception {
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.setBeanName("testRollbackNoRetries");
+		AtomicBoolean recovererShouldFail = new AtomicBoolean(true);
 		BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer = (rec, ex) -> {
-			throw new RuntimeException("arbp fail");
+			if (recovererShouldFail.getAndSet(false)) {
+				throw new RuntimeException("arbp fail");
+			}
 		};
 		DefaultAfterRollbackProcessor<Object, Object> afterRollbackProcessor =
 				spy(new DefaultAfterRollbackProcessor<>(recoverer, 0));

src/reference/asciidoc/kafka.adoc

@@ -2866,6 +2866,8 @@ This error handler does not support recovery, because the framework cannot know
 After seeking, an exception that wraps the `ListenerExecutionFailedException` is thrown.
 This is to cause the transaction to roll back (if transactions are enabled).
 
+IMPORTANT: If the recoverer fails (throws an exception), the record will be included in the seeks and recovery will be attempted again during the next delivery.
+
 ===== Container Stopping Error Handlers
 
 The `ContainerStoppingErrorHandler` (used with record listeners) stops the container if the listener throws an exception.
@@ -2916,6 +2918,8 @@ Starting with version 2.2.5, the `DefaultAfterRollbackProcessor` can be invoked
 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`.
 
+IMPORTANT: If the recoverer fails (throws an exception), the record will be included in the seeks and recovery will be attempted again during the next delivery.
+
 [[dead-letters]]
 ===== Publishing Dead-letter Records