diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index cad961104..f4d4ee74f 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -165,6 +165,7 @@ import org.springframework.util.backoff.FixedBackOff; * @author Byungjun You * @author Oliver Führer * @author Omer Celik + * @author Didier Loiseau */ public class KafkaMessageChannelBinder extends // @checkstyle:off @@ -566,10 +567,16 @@ public class KafkaMessageChannelBinder extends } @Override - @SuppressWarnings("unchecked") protected MessageProducer createConsumerEndpoint( final ConsumerDestination destination, final String group, final ExtendedConsumerProperties extendedConsumerProperties) { + return createConsumerEndpointCaptureHelper(destination, group, extendedConsumerProperties); + } + + @SuppressWarnings("unchecked") + private MessageProducer createConsumerEndpointCaptureHelper( + final ConsumerDestination destination, final String group, + final ExtendedConsumerProperties extendedConsumerProperties) { boolean anonymous = !StringUtils.hasText(group); Assert.isTrue( @@ -577,7 +584,7 @@ public class KafkaMessageChannelBinder extends "DLQ support is not available for anonymous subscriptions"); String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group; - final ConsumerFactory consumerFactory = createKafkaConsumerFactory( + final ConsumerFactory consumerFactory = (ConsumerFactory) createKafkaConsumerFactory( anonymous, consumerGroup, extendedConsumerProperties, destination.getName() + ".consumer", destination.getName()); int partitionCount = extendedConsumerProperties.getInstanceCount() * extendedConsumerProperties.getConcurrency(); @@ -647,9 +654,8 @@ public class KafkaMessageChannelBinder extends } resetOffsetsForAutoRebalance(extendedConsumerProperties, consumerFactory, containerProperties); containerProperties.setAuthExceptionRetryInterval(this.configurationProperties.getAuthorizationExceptionRetryInterval()); - @SuppressWarnings("rawtypes") - final ConcurrentMessageListenerContainer messageListenerContainer = new ConcurrentMessageListenerContainer( - consumerFactory, containerProperties) { + final ConcurrentMessageListenerContainer messageListenerContainer = + new ConcurrentMessageListenerContainer<>(consumerFactory, containerProperties) { @Override public void stop(Runnable callback) { @@ -677,8 +683,7 @@ public class KafkaMessageChannelBinder extends ContainerProperties.AckMode ackMode = extendedConsumerProperties.getExtension().getAckMode(); if (ackMode != null) { - if ((extendedConsumerProperties.isBatchMode() && ackMode != ContainerProperties.AckMode.RECORD) || - !extendedConsumerProperties.isBatchMode()) { + if (!extendedConsumerProperties.isBatchMode() || ackMode != ContainerProperties.AckMode.RECORD) { messageListenerContainer.getContainerProperties() .setAckMode(ackMode); } @@ -715,7 +720,7 @@ public class KafkaMessageChannelBinder extends } } else if (!extendedConsumerProperties.isBatchMode() && transMan != null) { - messageListenerContainer.setAfterRollbackProcessor(new DefaultAfterRollbackProcessor<>( + var afterRollbackProcessor = new DefaultAfterRollbackProcessor( (record, exception) -> { MessagingException payload = new MessagingException(((RecordMessageConverter) messageConverter) @@ -740,7 +745,31 @@ public class KafkaMessageChannelBinder extends } }, createBackOff(extendedConsumerProperties), new KafkaTemplate<>(transMan.getProducerFactory()), - extendedConsumerProperties.getExtension().isTxCommitRecovered())); + extendedConsumerProperties.getExtension().isTxCommitRecovered()); + if (!CollectionUtils.isEmpty(extendedConsumerProperties.getRetryableExceptions())) { + // mimic AbstractBinder.buildRetryTemplate(properties)’s retryPolicy + if (!extendedConsumerProperties.isDefaultRetryable()) { + afterRollbackProcessor.defaultFalse(true); + } + extendedConsumerProperties.getRetryableExceptions() + .forEach((t, retry) -> { + if (Exception.class.isAssignableFrom(t)) { + var ex = t.asSubclass(Exception.class); + if (retry) { + afterRollbackProcessor.addRetryableExceptions(ex); + } + else { + afterRollbackProcessor.addNotRetryableExceptions(ex); + } + } + else { + throw new IllegalArgumentException( + "Only Exception types can be configured as retryable-exceptions together with transactions. " + + "Unsupported type: " + t.getName()); + } + }); + } + messageListenerContainer.setAfterRollbackProcessor(afterRollbackProcessor); } else { kafkaMessageDrivenChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel()); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 06aced054..5157920fc 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -36,6 +36,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; import java.util.stream.IntStream; import com.fasterxml.jackson.databind.ObjectMapper; @@ -172,7 +173,8 @@ import static org.mockito.Mockito.verify; * @author Henryk Konsek * @author Gary Russell * @author Chris Bono - * @Author Oliver Führer + * @author Oliver Führer + * @author Didier Loiseau */ @EmbeddedKafka(count = 1, controlledShutdown = true, topics = "error.pollableDlq.group-pcWithDlq", brokerProperties = {"transaction.state.log.replication.factor=1", "transaction.state.log.min.isr=1"}) @@ -1052,11 +1054,41 @@ class KafkaBinderTests extends testDlqGuts(true, null, null, false, false); } + @Test + void dlqAndRetryWithNonRetryableException() throws Exception { + testDlqGuts(true, null, null, false, false, true, true); + } + + @Test + void dlqAndRetryDefaultFalse() throws Exception { + testDlqGuts(true, null, null, false, false, false, false); + } + + @Test + void dlqAndRetryDefaultFalseWithRetryableException() throws Exception { + testDlqGuts(true, null, null, false, false, false, true); + } + @Test void dlqAndRetryTransactional() throws Exception { testDlqGuts(true, null, null, true, false); } + @Test + void dlqAndRetryWithNonRetryableExceptionTransactional() throws Exception { + testDlqGuts(true, null, null, true, false, true, true); + } + + @Test + void dlqAndRetryDefaultFalseTransactional() throws Exception { + testDlqGuts(true, null, null, true, false, false, false); + } + + @Test + void dlqAndRetryDefaultFalseWithRetryableExceptionTransactional() throws Exception { + testDlqGuts(true, null, null, true, false, false, true); + } + @Test void dlq() throws Exception { testDlqGuts(false, null, 3, false, false); @@ -1084,6 +1116,14 @@ class KafkaBinderTests extends private void testDlqGuts(boolean withRetry, HeaderMode headerMode, Integer dlqPartitions, boolean transactional, boolean useDlqDestResolver) throws Exception { + testDlqGuts(withRetry, headerMode, dlqPartitions, transactional, + useDlqDestResolver, true, false); + } + + private void testDlqGuts(boolean withRetry, HeaderMode headerMode, + Integer dlqPartitions, boolean transactional, boolean useDlqDestResolver, + boolean defaultRetryable, boolean useConfiguredRetryableException) + throws Exception { int expectedDlqPartition = dlqPartitions == null ? 0 : dlqPartitions - 1; KafkaBinderConfigurationProperties binderConfig = createConfigurationProperties(); @@ -1128,12 +1168,18 @@ class KafkaBinderTests extends consumerProperties.getExtension().setDlqPartitions(dlqPartitions); consumerProperties.setConcurrency(2); consumerProperties.populateBindingName("foobar"); + consumerProperties.setDefaultRetryable(defaultRetryable); + consumerProperties.getRetryableExceptions().put(NumberFormatException.class, + !defaultRetryable); DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); var dlqChannel = new QueueChannel(); - var handler = new FailingInvocationCountingMessageHandler(); + var handler = new FailingInvocationCountingMessageHandler( + () -> useConfiguredRetryableException + ? new NumberFormatException("fail") + : new RuntimeException("fail")); moduleInputChannel.subscribe(handler); long uniqueBindingId = System.currentTimeMillis(); @@ -1253,8 +1299,10 @@ class KafkaBinderTests extends .get(KafkaHeaders.RECEIVED_PARTITION)).isEqualTo(expectedDlqPartition); } else if (!HeaderMode.none.equals(headerMode)) { + boolean shouldHaveRetried = defaultRetryable != useConfiguredRetryableException; assertThat(handler.getInvocationCount()) - .isEqualTo(consumerProperties.getMaxAttempts()); + .isEqualTo( + shouldHaveRetried ? consumerProperties.getMaxAttempts() : 1); assertThat(receivedMessage.getHeaders() .get(KafkaMessageChannelBinder.X_ORIGINAL_TOPIC)) @@ -4090,14 +4138,27 @@ class KafkaBinderTests extends private final class FailingInvocationCountingMessageHandler implements MessageHandler { + private final Supplier exceptionProvider; + private volatile int invocationCount; private final LinkedHashMap> receivedMessages = new LinkedHashMap<>(); private final CountDownLatch latch; - private FailingInvocationCountingMessageHandler(int latchSize) { + private FailingInvocationCountingMessageHandler(int latchSize, + Supplier exceptionProvider) { latch = new CountDownLatch(latchSize); + this.exceptionProvider = exceptionProvider; + } + + private FailingInvocationCountingMessageHandler( + Supplier exceptionProvider) { + this(1, exceptionProvider); + } + + private FailingInvocationCountingMessageHandler(int latchSize) { + this(latchSize, () -> new RuntimeException("fail")); } private FailingInvocationCountingMessageHandler() { @@ -4115,7 +4176,7 @@ class KafkaBinderTests extends receivedMessages.put(offset, message); latch.countDown(); } - throw new RuntimeException("fail"); + throw exceptionProvider.get(); } public LinkedHashMap> getReceivedMessages() {