diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index d72d516b3..34e8b2092 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -102,6 +102,7 @@ import org.springframework.kafka.listener.ConsumerAwareRebalanceListener; import org.springframework.kafka.listener.ConsumerProperties; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.DefaultAfterRollbackProcessor; +import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.ProducerListener; @@ -112,6 +113,7 @@ import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.transaction.KafkaAwareTransactionManager; import org.springframework.kafka.transaction.KafkaTransactionManager; import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; @@ -214,6 +216,8 @@ public class KafkaMessageChannelBinder extends private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties(); + private Map ackModeInfo = new ConcurrentHashMap<>(); + public KafkaMessageChannelBinder( KafkaBinderConfigurationProperties configurationProperties, KafkaTopicProvisioner provisioningProvider) { @@ -695,6 +699,7 @@ public class KafkaMessageChannelBinder extends kafkaMessageDrivenChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel()); } this.getContainerCustomizer().configure(messageListenerContainer, destination.getName(), group); + this.ackModeInfo.put(destination, messageListenerContainer.getContainerProperties().getAckMode()); return kafkaMessageDrivenChannelAdapter; } @@ -1172,16 +1177,31 @@ public class KafkaMessageChannelBinder extends String dlqName = StringUtils.hasText(kafkaConsumerProperties.getDlqName()) ? kafkaConsumerProperties.getDlqName() : "error." + record.topic() + "." + group; + MessageHeaders headers; + if (message instanceof ErrorMessage) { + final ErrorMessage errorMessage = (ErrorMessage) message; + final Message originalMessage = errorMessage.getOriginalMessage(); + if (originalMessage != null) { + headers = originalMessage.getHeaders(); + } + else { + headers = message.getHeaders(); + } + } + else { + headers = message.getHeaders(); + } if (this.transactionTemplate != null) { Throwable throwable2 = throwable; this.transactionTemplate.executeWithoutResult(status -> { dlqSender.sendToDlq(recordToSend.get(), kafkaHeaders, dlqName, group, throwable2, - determinDlqPartitionFunction(properties.getExtension().getDlqPartitions())); + determinDlqPartitionFunction(properties.getExtension().getDlqPartitions()), + headers, this.ackModeInfo.get(destination)); }); } else { dlqSender.sendToDlq(recordToSend.get(), kafkaHeaders, dlqName, group, throwable, - determinDlqPartitionFunction(properties.getExtension().getDlqPartitions())); + determinDlqPartitionFunction(properties.getExtension().getDlqPartitions()), headers, this.ackModeInfo.get(destination)); } }; } @@ -1453,7 +1473,8 @@ public class KafkaMessageChannelBinder extends @SuppressWarnings("unchecked") void sendToDlq(ConsumerRecord consumerRecord, Headers headers, - String dlqName, String group, Throwable throwable, DlqPartitionFunction partitionFunction) { + String dlqName, String group, Throwable throwable, DlqPartitionFunction partitionFunction, + MessageHeaders messageHeaders, ContainerProperties.AckMode ackMode) { K key = (K) consumerRecord.key(); V value = (V) consumerRecord.value(); ProducerRecord producerRecord = new ProducerRecord<>(dlqName, @@ -1483,6 +1504,9 @@ public class KafkaMessageChannelBinder extends KafkaMessageChannelBinder.this.logger .debug("Sent to DLQ " + sb.toString()); } + if (ackMode == ContainerProperties.AckMode.MANUAL || ackMode == ContainerProperties.AckMode.MANUAL_IMMEDIATE) { + messageHeaders.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class).acknowledge(); + } } }); } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index fdb8dfdfc..3fc339754 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -1124,7 +1124,6 @@ public class KafkaBinderTests extends consumerProperties.setBackOffInitialInterval(100); consumerProperties.setBackOffMaxInterval(150); consumerProperties.getExtension().setEnableDlq(true); - consumerProperties.getExtension().setAutoRebalanceEnabled(false); DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); @@ -1186,6 +1185,87 @@ public class KafkaBinderTests extends producerBinding.unbind(); } + //See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/870 for motivation for this test. + @Test + @SuppressWarnings("unchecked") + public void testAutoCommitOnErrorWhenManualAcknowledgement() throws Exception { + Binder binder = getBinder(); + ExtendedProducerProperties producerProperties = createProducerProperties(); + BindingProperties producerBindingProperties = createProducerBindingProperties( + producerProperties); + + DirectChannel moduleOutputChannel = createBindableChannel("output", + producerBindingProperties); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setMaxAttempts(3); + consumerProperties.setBackOffInitialInterval(100); + consumerProperties.setBackOffMaxInterval(150); + //When auto commit is disabled, then the record is committed after publishing to DLQ using the manual acknowledgement. + // (if DLQ is enabled, which is, in this case). + consumerProperties.getExtension().setAutoCommitOffset(false); + consumerProperties.getExtension().setEnableDlq(true); + + DirectChannel moduleInputChannel = createBindableChannel("input", + createConsumerBindingProperties(consumerProperties)); + + FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler(); + moduleInputChannel.subscribe(handler); + long uniqueBindingId = System.currentTimeMillis(); + Binding producerBinding = binder.bindProducer( + "retryTest." + uniqueBindingId + ".0", moduleOutputChannel, + producerProperties); + Binding consumerBinding = binder.bindConsumer( + "retryTest." + uniqueBindingId + ".0", "testGroup", moduleInputChannel, + consumerProperties); + ExtendedConsumerProperties dlqConsumerProperties = createConsumerProperties(); + dlqConsumerProperties.setMaxAttempts(1); + QueueChannel dlqChannel = new QueueChannel(); + Binding dlqConsumerBinding = binder.bindConsumer( + "error.retryTest." + uniqueBindingId + ".0.testGroup", null, dlqChannel, + dlqConsumerProperties); + + String testMessagePayload = "test." + UUID.randomUUID().toString(); + Message testMessage = MessageBuilder + .withPayload(testMessagePayload.getBytes()).build(); + moduleOutputChannel.send(testMessage); + + Message dlqMessage = receive(dlqChannel, 3); + assertThat(dlqMessage).isNotNull(); + assertThat(dlqMessage.getPayload()).isEqualTo(testMessagePayload.getBytes()); + + // first attempt fails + assertThat(handler.getReceivedMessages().entrySet()).hasSize(1); + Message handledMessage = handler.getReceivedMessages().entrySet().iterator() + .next().getValue(); + assertThat(handledMessage).isNotNull(); + assertThat( + new String((byte[]) handledMessage.getPayload(), StandardCharsets.UTF_8)) + .isEqualTo(testMessagePayload); + assertThat(handler.getInvocationCount()) + .isEqualTo(consumerProperties.getMaxAttempts()); + binderBindUnbindLatency(); + dlqConsumerBinding.unbind(); + consumerBinding.unbind(); + + // on the second attempt the message is not redelivered because the DLQ is set and the record in error is already committed. + QueueChannel successfulInputChannel = new QueueChannel(); + consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0", + "testGroup", successfulInputChannel, consumerProperties); + String testMessage2Payload = "test1." + UUID.randomUUID().toString(); + Message testMessage2 = MessageBuilder + .withPayload(testMessage2Payload.getBytes()).build(); + moduleOutputChannel.send(testMessage2); + + Message receivedMessage = receive(successfulInputChannel); + assertThat(receivedMessage.getPayload()) + .isEqualTo(testMessage2Payload.getBytes()); + + binderBindUnbindLatency(); + consumerBinding.unbind(); + producerBinding.unbind(); + } + @Test @SuppressWarnings("unchecked") public void testConfigurableDlqName() throws Exception {