SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1084: Add txCommitRecovered Property

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1084

Update copyrights
This commit is contained in:
Gary Russell
2021-06-02 16:12:57 -04:00
committed by Soby Chacko
parent 5adeea2acb
commit 7bc90c10a2
4 changed files with 41 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
docs/src/main/asciidoc/overview.adoc
View File

@@ -316,6 +316,11 @@ Usually needed if you want to synchronize another transaction with the Kafka tra
To achieve exactly once consumption and production of records, the consumer and producer bindings must all be configured with the same transaction manager.
+
Default: none.
txCommitRecovered::
When using a transactional binder, the offset of a recovered record (e.g. when retries are exhausted and the record is sent to a dead letter topic) will be committed via a new transaction, by default.
Setting this property to `false` suppresses committing the offset of recovered record.
+
Default: true.
[[reset-offsets]]
==== Resetting Offsets

15
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2021 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.
@@ -205,6 +205,11 @@ public class KafkaConsumerProperties {
*/
private String transactionManager;
/**
* Set to false to NOT commit the offset of a successfully recovered recovered in the after rollback processor.
*/
private boolean txCommitRecovered = true;
/**
* @return if each record needs to be acknowledged.
*
@@ -516,4 +521,12 @@ public class KafkaConsumerProperties {
this.transactionManager = transactionManager;
}
public boolean isTxCommitRecovered() {
return this.txCommitRecovered;
}
public void setTxCommitRecovered(boolean txCommitRecovered) {
this.txCommitRecovered = txCommitRecovered;
}
}

4
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -766,7 +766,9 @@ public class KafkaMessageChannelBinder extends
throw e;
}
}
}, createBackOff(extendedConsumerProperties)));
}, createBackOff(extendedConsumerProperties),
new KafkaTemplate<>(transMan.getProducerFactory()),
extendedConsumerProperties.getExtension().isTxCommitRecovered()));
}
else {
kafkaMessageDrivenChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel());

23
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2021 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.
@@ -121,7 +121,6 @@ import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaderMapper;
import org.springframework.kafka.support.KafkaHeaders;
@@ -1049,10 +1048,16 @@ public class KafkaBinderTests extends
Binding<MessageChannel> consumerBinding = binder.bindConsumer(consumerDest,
"testGroup", moduleInputChannel, consumerProperties);
MessageListenerContainer container = TestUtils.getPropertyValue(consumerBinding,
"lifecycle.messageListenerContainer", MessageListenerContainer.class);
AbstractMessageListenerContainer container = TestUtils.getPropertyValue(consumerBinding,
"lifecycle.messageListenerContainer", AbstractMessageListenerContainer.class);
assertThat(container.getContainerProperties().getTopicPartitionsToAssign().length)
.isEqualTo(4); // 2 topics 2 partitions each
if (transactional) {
assertThat(TestUtils.getPropertyValue(container.getAfterRollbackProcessor(), "kafkaTemplate")).isNotNull();
assertThat(
TestUtils.getPropertyValue(container.getAfterRollbackProcessor(), "commitRecovered", Boolean.class))
.isTrue();
}
String dlqTopic = useDlqDestResolver ? "foo.dlq" : "error.dlqTest." + uniqueBindingId + ".0.testGroup";
try (AdminClient admin = AdminClient.create(Collections.singletonMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
@@ -1074,6 +1079,7 @@ public class KafkaBinderTests extends
ExtendedConsumerProperties<KafkaConsumerProperties> dlqConsumerProperties = createConsumerProperties();
dlqConsumerProperties.setMaxAttempts(1);
dlqConsumerProperties.setHeaderMode(headerMode);
dlqConsumerProperties.getExtension().setTxCommitRecovered(false);
ApplicationContext context = TestUtils.getPropertyValue(binder.getBinder(),
"applicationContext", ApplicationContext.class);
@@ -1098,6 +1104,15 @@ public class KafkaBinderTests extends
dlqTopic, null, dlqChannel,
dlqConsumerProperties);
binderBindUnbindLatency();
if (transactional) {
assertThat(TestUtils.getPropertyValue(dlqConsumerBinding,
"lifecycle.messageListenerContainer.afterRollbackProcessor.kafkaTemplate")).isNotNull();
assertThat(
TestUtils.getPropertyValue(dlqConsumerBinding,
"lifecycle.messageListenerContainer.afterRollbackProcessor.commitRecovered", Boolean.class))
.isFalse();
}
String testMessagePayload = "test." + UUID.randomUUID().toString();
Message<byte[]> testMessage = MessageBuilder
.withPayload(testMessagePayload.getBytes())