diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AfterRollbackProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AfterRollbackProcessor.java index 9fe5cedf..8cfceaff 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AfterRollbackProcessor.java +++ b/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 { * 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> records, Consumer consumer, Exception exception, boolean recoverable); @@ -61,4 +67,17 @@ public interface AfterRollbackProcessor { // 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; + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java index a8f2c69c..5d02ae83 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java +++ b/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 outRecord, KafkaOperations 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 outRecord, KafkaOperations kafkaTemplate) { try { - template.send(outRecord).addCallback(result -> { + kafkaTemplate.send(outRecord).addCallback(result -> { if (logger.isDebugEnabled()) { logger.debug("Successful dead-letter publication: " + result); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java index 32aae467..74ffb6bf 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java +++ b/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 implements AfterRollbackProcess private final FailedRecordTracker failureTracker; + private boolean processInTransaction; + + private KafkaTemplate 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 implements AfterRollbackProcess @Override public void process(List> records, Consumer 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 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 kafkaTemplate) { + this.kafkaTemplate = kafkaTemplate; } @Override diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 7ee96de8..9fb31425 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -573,14 +573,14 @@ public class KafkaMessageListenerContainer // NOSONAR comment density } } - private void subscribeOrAssignTopics(final Consumer consumer) { + private void subscribeOrAssignTopics(final Consumer 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 // 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 // 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 // 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 // NOSONAR comment density try { ackImmediate(record); } - catch (WakeupException e) { + catch (@SuppressWarnings("unused") WakeupException e) { // ignore - not polling } } @@ -950,6 +950,7 @@ public class KafkaMessageListenerContainer // NOSONAR comment density @SuppressWarnings({ UNCHECKED, RAW_TYPES }) private void invokeBatchListenerInTx(final ConsumerRecords records, final List> recordList) { + try { this.transactionTemplate.execute(new TransactionCallbackWithoutResult() { @@ -972,15 +973,34 @@ public class KafkaMessageListenerContainer // NOSONAR comment density this.logger.error("Transaction rolled back", e); AfterRollbackProcessor afterRollbackProcessorToUse = (AfterRollbackProcessor) 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 records, + final List> recordList, RuntimeException e, + AfterRollbackProcessor afterRollbackProcessorToUse) { + + if (recordList == null) { + afterRollbackProcessorToUse.process(createRecordList(records), this.consumer, e, false); + } + else { + afterRollbackProcessorToUse.process(recordList, this.consumer, e, false); + } + } + private List> createRecordList(final ConsumerRecords records) { return StreamSupport.stream(records.spliterator(), false) .collect(Collectors.toList()); @@ -1020,7 +1040,7 @@ public class KafkaMessageListenerContainer // 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 // 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 records) { Iterator> iterator = records.iterator(); while (iterator.hasNext()) { @@ -1132,13 +1152,7 @@ public class KafkaMessageListenerContainer // NOSONAR comment density } catch (RuntimeException e) { this.logger.error("Transaction rolled back", e); - List> unprocessed = new ArrayList<>(); - unprocessed.add(record); - while (iterator.hasNext()) { - unprocessed.add(iterator.next()); - } - ((AfterRollbackProcessor) getAfterRollbackProcessor()) - .process(unprocessed, this.consumer, e, true); + recordAfterRollback(iterator, record, e); } finally { TransactionSupport.clearTransactionIdSuffix(); @@ -1146,6 +1160,32 @@ public class KafkaMessageListenerContainer // NOSONAR comment density } } + private void recordAfterRollback(Iterator> iterator, final ConsumerRecord record, + RuntimeException e) { + + List> unprocessed = new ArrayList<>(); + unprocessed.add(record); + while (iterator.hasNext()) { + unprocessed.add(iterator.next()); + } + @SuppressWarnings(UNCHECKED) + AfterRollbackProcessor afterRollbackProcessorToUse = + (AfterRollbackProcessor) 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 records) { Iterator> iterator = records.iterator(); while (iterator.hasNext()) { @@ -1499,7 +1539,7 @@ public class KafkaMessageListenerContainer // 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"); } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java index 290c43c9..0e3ad4c7 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java +++ b/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 container = new KafkaMessageListenerContainer<>(cf, containerProps); container.setBeanName("testMaxFailures"); final CountDownLatch recoverLatch = new CountDownLatch(1); - DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template) { + final KafkaTemplate 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 afterRollbackProcessor = + DefaultAfterRollbackProcessor 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"); } diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 9683b945..de138c16 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -2834,6 +2834,10 @@ In such cases, the application listener must handle a record that keeps failing. See also <>. +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