Is Non Blocking Retry (@Retryable) intended to be combined with Kafka Transactions (consumer initiated) ?

[PaulFert]

Discussed in #2918

^{Originally posted by PaulFert December 1, 2023}
Hello,

I wanted to use both features for consume -> process -> produce applications

When i use Transaction with Blocking Retry, DLT Recover the flow is :

When exception is raised from the listener, the transaction is rollback and then a new transaction is created for the send to DLT, OK fine.

On the oher side, when i use Transaction with Non Blocking Retry the flow is more like that :

When exception is raised from the listener, the transaction is not rollback, the message is send to retry topic and finally the transaction is comited. Then consumer offsets is commit plus the messages out and messages to retry topic. If an exception is raised i want my messages out to be not comited.

It's not what i expected, i expected a flow like that :

Did i miss something with the configuration ? I use standard configuration, with one kafkatemplate, with transactional producer
Or maybe did i misunderstood the Non Blocking Retry pattern, Transactions are not intended to be combined with ?

Sorry for my poor english...

Testing with 'org.springframework.boot' version '3.1.1-SNAPSHOT', spring-kafka version 3.0.8

Thanks

[Wzy19930507]

KafkaListener nested producer transactions are not support @RetryableTopic now.

---

Use @RetryableTopic, DeadLetterPublishingRecoverer will inject CommonErrorHandler.
RuntimeException catch by CommonErrorHandler but not throw RuntimeException again then transaction can not trigger rollback.

1. exception catch by CommonErrorHandler
   

   spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

   Lines 2889 to 2907 in e7e77b9

   	private void invokeErrorHandler(final ConsumerRecord<K, V> cRecord,
   	Iterator<ConsumerRecord<K, V>> iterator, RuntimeException rte) {
   	if (this.commonErrorHandler.seeksAfterHandling() || rte instanceof CommitFailedException) {
   	try {
   	if (this.producer == null) {
   	processCommits();
   	}
   	}
   	catch (Exception ex) { // NO SONAR
   	this.logger.error(ex, "Failed to commit before handling error");
   	}
   	List<ConsumerRecord<?, ?>> records = new ArrayList<>();
   	records.add(cRecord);
   	while (iterator.hasNext()) {
   	records.add(iterator.next());
   	}
   	this.commonErrorHandler.handleRemaining(rte, records, this.consumer,
   	KafkaMessageListenerContainer.this.thisOrParentContainer);

2. CommonErrorHandler call DeadLetterPublishingRecoverer.publish, only log not threw again.
   

   spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

   Lines 684 to 706 in e7e77b9

   	protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate,
   	ConsumerRecord<?, ?> inRecord) {
   	CompletableFuture<SendResult<Object, Object>> sendResult = null;
   	try {
   	sendResult = kafkaTemplate.send(outRecord);
   	sendResult.whenComplete((result, ex) -> {
   	if (ex == null) {
   	this.logger.debug(() -> "Successful dead-letter publication: "
   	+ KafkaUtils.format(inRecord) + " to " + result.getRecordMetadata());
   	}
   	else {
   	this.logger.error(ex, () -> pubFailMessage(outRecord, inRecord));
   	}
   	});
   	}
   	catch (Exception e) {
   	this.logger.error(e, () -> pubFailMessage(outRecord, inRecord));
   	}
   	if (this.failIfSendResultIsError) {
   	verifySendResult(kafkaTemplate, outRecord, sendResult, inRecord);
   	}
   	}

---

If not use @RetryTopic, will throw exception to transaction and rollback, Then AfterRollbackProcessor catch the exception.
AfterRollbackProcessor default injected DeadLetterPublishingRecoverer.

[Wzy19930507]

Hi, @artembilan @sobychacko @PaulFert for this scenario(enable listener transaction and retry topic), retry topic break the listener kafka transaction atomicity, move retry topic logic to AfterRollbackProcessor feelings can be resolved this question.

WDYT?