Skip to content

Commit d3816ac

Browse files
committed
Non-blocking retries will break container transactions atomicity.
* Enable non-blocking retries and container transaction, when listener code threw error, catch by `DefaultErrorHandler` and recover success. Container transactions commit success and send record to a retryable topic. * Desired behavior is Container transactions rollback, send offsets commit transaction success, send record to a retryable topic. * Add adoc to notice. see #2934
1 parent ac39bc2 commit d3816ac

File tree

3 files changed

+5
-325
lines changed

3 files changed

+5
-325
lines changed

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/transactions.adoc

Lines changed: 3 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,9 @@ Instead, use a `KafkaTransactionManager` in the container to start the Kafka tra
7979

8080
See xref:tips.adoc#ex-jdbc-sync[Examples of Kafka Transactions with Other Transaction Managers] for an example application that chains JDBC and Kafka transactions.
8181

82+
IMPORTANT: Container transactions use xref:retrytopic.adoc[Non-Blocking Retries] will break transaction atomicity.
83+
When listener code threw exception, container transactions commit success and record sending to the retryable topic.
84+
8285
[[kafkatemplate-local-transactions]]
8386
== `KafkaTemplate` Local Transactions
8487

@@ -209,67 +212,3 @@ public static class Config {
209212
}
210213
----
211214

212-
[[kafka-listener-nested-kafka-transactions]]
213-
== Listeners Nested Kafka Transactions
214-
215-
IMPORTANT: xref:retrytopic.adoc[Non-Blocking Retries] are not supported Listeners Nested Kafka transactions.
216-
Listeners enable transaction and Non-Blocking Retries, when listener fails after invoke `KafkaTemplate` Transactions.
217-
`KafkaTemplate` transactions commit success multiple times because of exception catch by Non-Blocking Retries and not threw.
218-
219-
[source, java]
220-
----
221-
public static class TestListener {
222-
223-
@Autowired
224-
KafkaTemplate<String, String> kafkaTemplate;
225-
226-
@RetryableTopic
227-
@KafkaListener(topics = "tests")
228-
void listen(String in) {
229-
kafkaTemplate.send("nested-test-topic", "value");
230-
throw new RuntimeException("from FirstRetryableKafkaListener");
231-
}
232-
233-
}
234-
235-
@EnableKafka
236-
@Configuration
237-
static class Config {
238-
239-
@Bean
240-
public TestListener test() {
241-
return new TestListener();
242-
}
243-
244-
@Bean
245-
public ConsumerFactory<?, ?> consumerFactory() {
246-
return mock(ConsumerFactory.class);
247-
}
248-
249-
@Bean
250-
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
251-
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
252-
factory.setConsumerFactory(consumerFactory());
253-
ContainerProperties props = factory.getContainerProperties();
254-
KafkaTransactionManager<String, String> tm = new KafkaTransactionManager<>(producerFactory());
255-
props.setTransactionManager(tm);
256-
return factory;
257-
}
258-
259-
@Bean
260-
ProducerFactory<String, String> producerFactory() {
261-
Map<String, Object> configProps = new HashMap<>();
262-
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(configProps);
263-
pf.setTransactionIdPrefix("listener.noRetries.");
264-
return pf;
265-
}
266-
267-
@Bean
268-
KafkaTemplate<String, String> kafkaTemplate() {
269-
KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
270-
kafkaTemplate.setConsumerFactory(consumerFactory());
271-
return kafkaTemplate;
272-
}
273-
274-
}
275-
----

spring-kafka-docs/src/main/antora/modules/ROOT/pages/retrytopic.adoc

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,4 +9,5 @@ Since 2.7 Spring for Apache Kafka offers support for that via the `@RetryableTop
99

1010
IMPORTANT: Non-blocking retries are not supported with xref:kafka/receiving-messages/listener-annotation.adoc#batch-listeners[Batch Listeners].
1111

12-
IMPORTANT: Non-blocking retries are not supported xref:kafka/transactions.adoc#kafka-listener-nested-kafka-transactions[Listeners Nested Kafka Transactions].
12+
13+
IMPORTANT: xref:kafka/transactions.adoc#container-transaction-manager[Using Consumer-Initiated Transactions] use Non-Blocking Retries will break transaction atomicity.

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicTransactionIntegrationTests.java

Lines changed: 0 additions & 260 deletions
This file was deleted.

0 commit comments

Comments
 (0)