This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new aebc65584810 CAMEL-26856: camel-kafka - exactlyOnce: a rejected send
whose failure a doCatch caught skips the record, as for an error handler
(#29667)
aebc65584810 is described below
commit aebc655848102d633ff446c4abcd3747fa03f482
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Nov 12 17:33:12 2026 +0530
CAMEL-26856: camel-kafka - exactlyOnce: a rejected send whose failure a
doCatch caught skips the record, as for an error handler (#29667)
* CAMEL-26856: camel-kafka - exactlyOnce: a rejected send whose failure a
doCatch caught skips the record, as for an error handler
With exactlyOnce, once a Kafka send failed in the open transaction, the
next send through the source producer (a redelivery of the failed send, or
another send, such as in a loop) is rejected: KafkaProducer marks the
exchange rollback only, sets the retry rollback marker and throws.
KafkaTransactionSynchronization took the offsets-only path (abort the
output, commit the source offsets, skip the record) for such a rejection
only when the exchange was failure handled, that is when an error handler
handled it. A doCatch does not mark the exchange failure handled when the
doTry failed in its own route, and CAMEL-26447 (not merged yet) stops it
when the doTry called the route of the send. Without that mark the source
transaction was aborted and the record replayed (for the doTry in the
route of the send today, and for the calling route once CAMEL-26447 is
merged); a send that keeps failing then never gets past the record.
A rejected send now counts as handled when the exchange has no exception
left and the first send failure is recorded (or the exchange is failure
handled), whoever handled it. A rejection that nothing handled keeps its
exception and is still rolled back, and so is a rollback the route asks
for without the marker, or the marker without a recorded send failure.
The kafka component page says which sends are rejected, that a doCatch
that catches one skips the record, and how to replay it instead (throw
again, or rollback()); its example of maximumRedeliveries(2) is corrected
(the first rejected retry ends the redeliveries). The 4.24 upgrade guide
names the doCatch in the route of the send, which replayed the record
before.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-26856: camel-kafka - exactlyOnce: the consumer does not report a
record skipped after a doCatch as rolled back
Review of #29667:
- After a doCatch, CatchProcessor sets rollbackOnly again, so the consumer's
completion (KafkaRecordStreamingProcessor.completeExactlyOnceProcessing)
saw
a rollback-only exchange after the offsets-only commit, reported the
record
as rolled back and, with breakOnFirstError (which exactlyOnce requires),
reconnected. KafkaTransactionSynchronization now marks the source metadata
once the offsets-only transaction committed, and the consumer does not
treat
rollback only as a rollback when that mark is set and the transaction was
not
aborted. The rollback-only flag of the exchange is left as it is.
- The comment no longer says "once CAMEL-26447 is merged"; the condition is
named handledRejectedSend.
- KafkaExactlyOnceIT: a doTry calls a direct route with
maximumRedeliveries(1)
whose sink rejects the poison record: the record is skipped, not replayed,
and the consumer does not reconnect.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/kafka-component.adoc | 12 ++-
.../camel-kafka/src/main/docs/kafka-component.adoc | 12 ++-
.../kafka/KafkaTransactionSynchronization.java | 21 +++-
.../kafka/consumer/KafkaExactlyOnceMetadata.java | 18 ++++
.../streaming/KafkaRecordStreamingProcessor.java | 5 +-
.../kafka/KafkaTransactionSynchronizationTest.java | 110 +++++++++++++++++++++
.../KafkaRecordStreamingProcessorTest.java | 56 +++++++++++
.../kafka/integration/KafkaExactlyOnceIT.java | 56 +++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 ++
9 files changed, 290 insertions(+), 10 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc
index 853775f99ac5..65778e2a10ac 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc
@@ -688,8 +688,16 @@ steps without replaying those steps. `maximumRedeliveries`
may cause Camel to in
each in-transaction retry is rejected; the Kafka send is not retried in a
fresh transaction. After those rejected attempts,
a failure handler such as a `deadLetterChannel` or an
`onException(...).handled(true)` clause can handle the original
failure. When a supported non-Kafka failure-handler delivery succeeds, Camel
skips the source record using the offset-only
-transaction described below. With `maximumRedeliveries(2)`, for example, the
two retries are rejected and the
-failure-handler destination is invoked once; they are not two new Kafka send
attempts.
+transaction described below. With `maximumRedeliveries(2)`, for example, the
first retry is rejected, which marks the
+exchange rollback-only and ends the redeliveries, and the failure-handler
destination is invoked once; there is no new
+Kafka send attempt.
+
+Any later send through the source producer in that transaction is rejected the
same way, also one that is not a
+redelivery, such as a second send in a `loop` or in another `doTry` block.
When a `doCatch` catches such a rejected
+send, in the route of the send or in a route that called it from a `doTry`
(for example with `direct`), the failure is
+handled too, and Camel skips the source record using the offset-only
transaction. To replay the record instead, throw
+the exception again in the `doCatch` block, or end it with `rollback()`:
`markRollbackOnly()` is not told apart from
+the rejected send.
When a Kafka send failure, or another route failure, is handled successfully
while the source-output transaction is
already open, Camel aborts the source-output transaction and commits only the
source offsets in a new transaction. This
diff --git a/components/camel-kafka/src/main/docs/kafka-component.adoc
b/components/camel-kafka/src/main/docs/kafka-component.adoc
index 853775f99ac5..65778e2a10ac 100644
--- a/components/camel-kafka/src/main/docs/kafka-component.adoc
+++ b/components/camel-kafka/src/main/docs/kafka-component.adoc
@@ -688,8 +688,16 @@ steps without replaying those steps. `maximumRedeliveries`
may cause Camel to in
each in-transaction retry is rejected; the Kafka send is not retried in a
fresh transaction. After those rejected attempts,
a failure handler such as a `deadLetterChannel` or an
`onException(...).handled(true)` clause can handle the original
failure. When a supported non-Kafka failure-handler delivery succeeds, Camel
skips the source record using the offset-only
-transaction described below. With `maximumRedeliveries(2)`, for example, the
two retries are rejected and the
-failure-handler destination is invoked once; they are not two new Kafka send
attempts.
+transaction described below. With `maximumRedeliveries(2)`, for example, the
first retry is rejected, which marks the
+exchange rollback-only and ends the redeliveries, and the failure-handler
destination is invoked once; there is no new
+Kafka send attempt.
+
+Any later send through the source producer in that transaction is rejected the
same way, also one that is not a
+redelivery, such as a second send in a `loop` or in another `doTry` block.
When a `doCatch` catches such a rejected
+send, in the route of the send or in a route that called it from a `doTry`
(for example with `direct`), the failure is
+handled too, and Camel skips the source record using the offset-only
transaction. To replay the record instead, throw
+the exception again in the `doCatch` block, or end it with `rollback()`:
`markRollbackOnly()` is not told apart from
+the rejected send.
When a Kafka send failure, or another route failure, is handled successfully
while the source-output transaction is
already open, Camel aborts the source-output transaction and commits only the
source offsets in a new transaction. This
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaTransactionSynchronization.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaTransactionSynchronization.java
index e3a9012440af..07f6bf846aa5 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaTransactionSynchronization.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaTransactionSynchronization.java
@@ -121,8 +121,14 @@ class KafkaTransactionSynchronization extends
SynchronizationAdapter {
:
exchange.getProperty(KafkaExactlyOnceMetadata.SECONDARY_SEND_FAILURE_PROPERTY,
Throwable.class);
var exchangeExtension = exchange.getExchangeExtension();
boolean failureHandled = exchangeExtension != null &&
exchangeExtension.isFailureHandled();
- boolean handledRedeliveryRejection = failureHandled
- &&
Boolean.TRUE.equals(exchange.getProperty(KafkaExactlyOnceMetadata.RETRY_ROLLBACK_PROPERTY));
+ // a send that the producer rejected in the open transaction after
an earlier send failed (a redelivery, or
+ // another send through the producer, such as in a loop), which
marked the exchange rollback only, and whose
+ // failure was handled: by an error handler, or by a doCatch
(which leaves no exception but does not mark
+ // the exchange failure handled when the doTry failed in its own
route, or when it caught the failure of a
+ // route the doTry called). A rejection that nothing handled keeps
its exception and is rolled back below
+ boolean handledRejectedSend = exchange.getException() == null
+ &&
Boolean.TRUE.equals(exchange.getProperty(KafkaExactlyOnceMetadata.RETRY_ROLLBACK_PROPERTY))
+ && (failureHandled || sendFailure != null);
if (exchangeMetadata != null &&
exchangeMetadata.isClaimRejected()) {
if (exchange.getException() == null) {
@@ -140,7 +146,7 @@ class KafkaTransactionSynchronization extends
SynchronizationAdapter {
rollback(exchange);
markAborted(exchange);
} else if (exchange.getException() != null
- || (exchange.isRollbackOnly() &&
!handledRedeliveryRejection)) {
+ || (exchange.isRollbackOnly() && !handledRejectedSend)) {
rollback(exchange);
markAborted(exchange);
} else if (offsetsToCommit != null && sendFailure != null &&
secondarySendFailure != null) {
@@ -163,7 +169,7 @@ class KafkaTransactionSynchronization extends
SynchronizationAdapter {
if (sourceFailure == null) {
sourceFailure = new IllegalStateException("A handled route
failure completed without an exception");
}
- commitOffsetsAfterHandledFailure(exchange, sourceFailure);
+ commitOffsetsAfterHandledFailure(exchange, exchangeMetadata,
sourceFailure);
} else {
commit(exchange);
if (exchange.getException() != null) {
@@ -211,7 +217,8 @@ class KafkaTransactionSynchronization extends
SynchronizationAdapter {
}
}
- private void commitOffsetsAfterHandledFailure(Exchange exchange, Throwable
sourceFailure) {
+ private void commitOffsetsAfterHandledFailure(
+ Exchange exchange, KafkaExactlyOnceMetadata exchangeMetadata,
Throwable sourceFailure) {
try {
kafkaProducer.abortTransaction();
} catch (Exception abortFailure) {
@@ -235,6 +242,10 @@ class KafkaTransactionSynchronization extends
SynchronizationAdapter {
kafkaProducer.beginTransaction();
kafkaProducer.sendOffsetsToTransaction(offsetsToCommit,
groupMetadata);
kafkaProducer.commitTransaction();
+ if (exchangeMetadata != null) {
+ // the consumer must not report the skipped record as rolled
back when a doCatch left it rollback only
+ exchangeMetadata.markOffsetsCommittedAfterHandledFailure();
+ }
LOG.warn("Skipped source offset(s) in transaction {} after a
handled source failure: {}",
transactionId, sourceFailure.toString());
} catch (Exception offsetCommitFailure) {
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/KafkaExactlyOnceMetadata.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/KafkaExactlyOnceMetadata.java
index b9de875aaa2f..bdc3745bbc11 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/KafkaExactlyOnceMetadata.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/KafkaExactlyOnceMetadata.java
@@ -115,6 +115,7 @@ public final class KafkaExactlyOnceMetadata {
private final AtomicReference<Throwable> secondarySendFailure = new
AtomicReference<>();
private final AtomicBoolean deadLetterSendFailed = new AtomicBoolean();
private final AtomicBoolean asynchronousKafkaHandoff = new AtomicBoolean();
+ private final AtomicBoolean offsetsCommittedAfterHandledFailure = new
AtomicBoolean();
private void recordSendFailure(Throwable failure) {
Throwable firstFailure = sendFailure.get();
@@ -161,6 +162,23 @@ public final class KafkaExactlyOnceMetadata {
return asynchronousKafkaHandoff.get();
}
+ /**
+ * Mark that the transaction synchronization committed the source offsets
alone after a handled failure: the record
+ * is skipped, even when the exchange is still rollback only (a doCatch
restores the flag after it handled the
+ * failure).
+ */
+ public void markOffsetsCommittedAfterHandledFailure() {
+ offsetsCommittedAfterHandledFailure.set(true);
+ }
+
+ /**
+ * Whether the source offsets were committed alone after a handled failure
(the record is skipped). Set at most once
+ * per record (one source transaction commits its offsets), so {@link
#clearSendFailures()} does not reset it.
+ */
+ public boolean isOffsetsCommittedAfterHandledFailure() {
+ return offsetsCommittedAfterHandledFailure.get();
+ }
+
/** Clear failures that occurred before a new transaction synchronization
took ownership of the offsets. */
public void clearSendFailures() {
sendFailure.set(null);
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessor.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessor.java
index 5c90ba2bde41..ab2aa6afa5ea 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessor.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessor.java
@@ -191,7 +191,10 @@ final class KafkaRecordStreamingProcessor extends
KafkaRecordProcessor {
boolean transactionAborted = Boolean.TRUE.equals(
exchange.getProperty(KafkaExactlyOnceMetadata.TRANSACTION_ABORTED_PROPERTY));
- if (transactionAborted || exchange.isRollbackOnly()) {
+ // the transaction synchronization committed the source offsets alone
after a handled failure (the record is
+ // skipped): a doCatch restores rollbackOnly after it handled the
failure, which is not a rollback here
+ boolean skipped = metadata != null &&
metadata.isOffsetsCommittedAfterHandledFailure();
+ if (transactionAborted || (exchange.isRollbackOnly() && !skipped)) {
if (exchange.getException() == null) {
exchange.setException(new IllegalStateException(
"The Kafka exactlyOnce transaction was rolled back;
the source record must be replayed"));
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaTransactionSynchronizationTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaTransactionSynchronizationTest.java
index 0861b5b901b2..58e408692baf 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaTransactionSynchronizationTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaTransactionSynchronizationTest.java
@@ -37,6 +37,8 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -252,6 +254,105 @@ class KafkaTransactionSynchronizationTest {
verify(owner).releaseExactlyOnceTransaction("source-exchange");
}
+ @Test
+ void
redeliveryRejectionCaughtByDoCatchOfCallingRouteCommitsOffsetsAfterAbortingOutput()
{
+ // a send failed and the next send in the transaction was rejected
(its redelivery in a route called from a
+ // doTry, or a second send in the route), and a doCatch caught the
failure: no exception is left, but the
+ // exchange is not failure handled
+ when(exchange.getException()).thenReturn(null);
+ when(exchange.isRollbackOnly()).thenReturn(true);
+
when(exchange.getProperty(KafkaExactlyOnceMetadata.RETRY_ROLLBACK_PROPERTY)).thenReturn(Boolean.TRUE);
+ KafkaExactlyOnceMetadata metadata = sourceMetadataWithSendFailure();
+ Map<TopicPartition, OffsetAndMetadata> offsets =
metadata.getOffsetsToCommit();
+ ConsumerGroupMetadata group = metadata.getConsumerGroupMetadata();
+ sync = new KafkaTransactionSynchronization(TX_ID, kafkaProducer,
owner, metadata, "source-exchange");
+
+ sync.onDone(exchange);
+
+ var order = inOrder(kafkaProducer);
+ order.verify(kafkaProducer).abortTransaction();
+ order.verify(kafkaProducer).beginTransaction();
+ order.verify(kafkaProducer).sendOffsetsToTransaction(offsets, group);
+ order.verify(kafkaProducer).commitTransaction();
+ verify(kafkaProducer, never()).close();
+ verify(exchange, never()).setException(Mockito.any());
+ verify(exchange,
never()).setProperty(KafkaExactlyOnceMetadata.TRANSACTION_ABORTED_PROPERTY,
Boolean.TRUE);
+ verify(owner).releaseExactlyOnceTransaction("source-exchange");
+ // the consumer must not report the skipped record as rolled back
although the exchange is still rollback only
+ assertTrue(metadata.isOffsetsCommittedAfterHandledFailure());
+ }
+
+ @Test
+ void
rejectedSendCaughtByDoCatchIsNotMarkedSkippedWhenTheOffsetCommitFails() {
+ when(exchange.getException()).thenReturn(null);
+ when(exchange.isRollbackOnly()).thenReturn(true);
+
when(exchange.getProperty(KafkaExactlyOnceMetadata.RETRY_ROLLBACK_PROPERTY)).thenReturn(Boolean.TRUE);
+ KafkaExactlyOnceMetadata metadata = sourceMetadataWithSendFailure();
+ // the offset-only transaction fails at its commit: the record is
replayed, so it is not marked skipped
+ Mockito.doThrow(new KafkaException("offset commit
failed")).when(kafkaProducer).commitTransaction();
+ sync = new KafkaTransactionSynchronization(TX_ID, kafkaProducer,
owner, metadata, "source-exchange");
+
+ sync.onDone(exchange);
+
+ verify(kafkaProducer).sendOffsetsToTransaction(Mockito.anyMap(),
Mockito.any(ConsumerGroupMetadata.class));
+
verify(exchange).setProperty(KafkaExactlyOnceMetadata.TRANSACTION_ABORTED_PROPERTY,
Boolean.TRUE);
+ assertFalse(metadata.isOffsetsCommittedAfterHandledFailure());
+ }
+
+ @Test
+ void redeliveryRejectionMarkerWithoutRecordedSendFailureRollsBack() {
+ // control: the marker alone, with no recorded send failure and not
failure handled, does not commit the output
+ when(exchange.getException()).thenReturn(null);
+ when(exchange.isRollbackOnly()).thenReturn(true);
+
when(exchange.getProperty(KafkaExactlyOnceMetadata.RETRY_ROLLBACK_PROPERTY)).thenReturn(Boolean.TRUE);
+ KafkaExactlyOnceMetadata metadata = sourceMetadata();
+ sync = new KafkaTransactionSynchronization(TX_ID, kafkaProducer,
owner, metadata, "source-exchange");
+
+ sync.onDone(exchange);
+
+ verify(kafkaProducer).abortTransaction();
+ verify(kafkaProducer, never()).beginTransaction();
+ verify(kafkaProducer, never()).commitTransaction();
+ verify(owner).releaseExactlyOnceTransaction("source-exchange");
+ assertFalse(metadata.isOffsetsCommittedAfterHandledFailure());
+ }
+
+ @Test
+ void redeliveryRejectionThatNothingHandledRollsBack() {
+ // control: the rejected retry fails the exchange, nothing caught it
+ when(exchange.getException()).thenReturn(new
IllegalStateException("retry rejected"));
+ when(exchange.isRollbackOnly()).thenReturn(true);
+
when(exchange.getProperty(KafkaExactlyOnceMetadata.RETRY_ROLLBACK_PROPERTY)).thenReturn(Boolean.TRUE);
+ KafkaExactlyOnceMetadata metadata = sourceMetadataWithSendFailure();
+ sync = new KafkaTransactionSynchronization(TX_ID, kafkaProducer,
owner, metadata, "source-exchange");
+
+ sync.onDone(exchange);
+
+ verify(kafkaProducer).abortTransaction();
+ verify(kafkaProducer, never()).beginTransaction();
+ verify(kafkaProducer,
never()).sendOffsetsToTransaction(Mockito.anyMap(),
Mockito.any(ConsumerGroupMetadata.class));
+ verify(kafkaProducer, never()).commitTransaction();
+ verify(owner).releaseExactlyOnceTransaction("source-exchange");
+ }
+
+ @Test
+ void rollbackOnlyWithoutRetryMarkerAfterDoCatchRollsBack() {
+ // control: a route that asks for a rollback after a doCatch, without
a rejected retry, is rolled back
+ when(exchange.getException()).thenReturn(null);
+ when(exchange.isRollbackOnly()).thenReturn(true);
+
when(exchange.getProperty(KafkaExactlyOnceMetadata.RETRY_ROLLBACK_PROPERTY)).thenReturn(null);
+ KafkaExactlyOnceMetadata metadata = sourceMetadataWithSendFailure();
+ sync = new KafkaTransactionSynchronization(TX_ID, kafkaProducer,
owner, metadata, "source-exchange");
+
+ sync.onDone(exchange);
+
+ verify(kafkaProducer).abortTransaction();
+ verify(kafkaProducer, never()).beginTransaction();
+ verify(kafkaProducer,
never()).sendOffsetsToTransaction(Mockito.anyMap(),
Mockito.any(ConsumerGroupMetadata.class));
+ verify(kafkaProducer, never()).commitTransaction();
+ verify(owner).releaseExactlyOnceTransaction("source-exchange");
+ }
+
@Test
void
firstExactlyOnceKafkaDeadLetterProducerCommitsRecordAndSourceOffsetsTogether() {
when(exchange.getException()).thenReturn(null);
@@ -458,6 +559,15 @@ class KafkaTransactionSynchronizationTest {
verify(owner).markProducerClosedForRecreation();
}
+ private KafkaExactlyOnceMetadata sourceMetadataWithSendFailure() {
+ // the first send failure is recorded on the source metadata, as
KafkaProducer does before it rejects the retry
+ KafkaExactlyOnceMetadata metadata = sourceMetadata();
+ when(exchange.getProperty(KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
KafkaExactlyOnceMetadata.class))
+ .thenReturn(metadata);
+ KafkaExactlyOnceMetadata.recordSendFailure(exchange, new
KafkaException("the record is too large"));
+ return metadata;
+ }
+
private static KafkaExactlyOnceMetadata sourceMetadata() {
DefaultKafkaManualCommit manual =
Mockito.mock(DefaultKafkaManualCommit.class);
ConsumerGroupMetadata group =
Mockito.mock(ConsumerGroupMetadata.class);
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorTest.java
index d1617d06324e..c44a06ae7191 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorTest.java
@@ -31,9 +31,11 @@ import
org.apache.camel.component.kafka.consumer.CommitManager;
import org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommit;
import org.apache.camel.component.kafka.consumer.KafkaExactlyOnceMetadata;
import org.apache.camel.component.kafka.consumer.KafkaManualCommit;
+import org.apache.camel.component.kafka.consumer.support.ProcessingResult;
import org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.spi.HeaderFilterStrategy;
+import org.apache.camel.support.DefaultExchange;
import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
@@ -43,9 +45,13 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -92,6 +98,56 @@ class KafkaRecordStreamingProcessorTest {
verify(commitManager).recordOffset(topicPartition, 42L);
}
+ @Test
+ void
recordSkippedAfterAHandledFailureIsNotReportedAsRolledBackWhenStillRollbackOnly()
throws Exception {
+ // a doCatch handled the failure of a send, the transaction
synchronization aborted the output and committed the
+ // source offsets alone, and the doCatch restored rollbackOnly
+ Exchange exchange = new DefaultExchange(new DefaultCamelContext());
+
+ ProcessingResult result = processExactlyOnce(exchange, true);
+
+ assertNull(exchange.getException(), "the skipped record must not be
reported as rolled back");
+ assertFalse(result.isBreakOnErrorHit(), "the consumer must not break
(seek back and reconnect)");
+ assertFalse(result.isFailed());
+ verify(commitManager).recordOffset(topicPartition, 42L);
+ verify(commitManager, never()).commit(any(TopicPartition.class));
+ }
+
+ @Test
+ void rollbackOnlyRecordIsReportedAsRolledBack() throws Exception {
+ // control: rollbackOnly without the offset-only commit is a rollback,
and the record is replayed
+ Exchange exchange = new DefaultExchange(new DefaultCamelContext());
+
+ ProcessingResult result = processExactlyOnce(exchange, false);
+
+ assertInstanceOf(IllegalStateException.class, exchange.getException());
+ assertTrue(exchange.getException().getMessage().contains("rolled
back"), exchange.getException().getMessage());
+ assertTrue(result.isBreakOnErrorHit());
+ verify(commitManager, never()).recordOffset(topicPartition, 42L);
+ }
+
+ private ProcessingResult processExactlyOnce(Exchange exchange, boolean
offsetsCommittedAfterHandledFailure)
+ throws Exception {
+ // exactlyOnce needs breakOnFirstError
+ when(configuration.isExactlyOnce()).thenReturn(true);
+ when(configuration.isBreakOnFirstError()).thenReturn(true);
+ when(kafkaConsumer.createExchange(false)).thenReturn(exchange);
+ KafkaExactlyOnceMetadata metadata = newMetadata();
+ doAnswer(invocation -> {
+ Exchange e = invocation.getArgument(0);
+ e.setProperty(KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
metadata);
+ metadata.claim(e.getExchangeId());
+ if (offsetsCommittedAfterHandledFailure) {
+ metadata.markOffsetsCommittedAfterHandledFailure();
+ }
+ e.setRollbackOnly(true);
+ return null;
+ }).when(processor).process(any(Exchange.class));
+
+ KafkaRecordStreamingProcessor proc = new
KafkaRecordStreamingProcessor(configuration, processor, commitManager);
+ return proc.processExchange(kafkaConsumer, topicPartition, false,
false, record);
+ }
+
@ParameterizedTest
@ValueSource(strings = {
"seda:async-handoff-test",
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceIT.java
index ed85c0aa11b0..047b80dc49d9 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceIT.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceIT.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.kafka.integration;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -25,7 +26,10 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.CamelContext;
import org.apache.camel.builder.AggregationStrategies;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.DefaultKafkaClientFactory;
+import org.apache.camel.component.kafka.KafkaComponent;
import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.kafka.clients.consumer.Consumer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledOnOs;
import org.junit.jupiter.api.condition.OS;
@@ -545,6 +549,58 @@ class KafkaExactlyOnceIT extends KafkaExactlyOnceITSupport
{
"Both Kafka sends must be treated as normal sends in the
source transaction");
}
+ /**
+ * The send of big-1 fails in a route called from a doTry, its redelivery
is rejected in the open transaction, and
+ * the doCatch of the calling route handles the failure: the record is
skipped (not replayed), and the consumer does
+ * not report it as rolled back although the doCatch leaves the exchange
rollback only. With exactlyOnce the
+ * consumer breaks on the first error, so such a report would show as a
reconnect: a second Kafka consumer.
+ */
+ @Test
+ void rejectedRedeliveryCaughtByDoCatchOfCallingRouteSkipsTheRecord()
throws Exception {
+ String in = topics("in", "out")[0];
+ String publishUri = "direct:" + id + "-publish";
+ AtomicInteger caught = new AtomicInteger();
+ AtomicInteger consumersCreated = new AtomicInteger();
+ CamelContext context = newContext();
+ context.getComponent("kafka",
KafkaComponent.class).setKafkaClientFactory(new DefaultKafkaClientFactory() {
+ @Override
+ public Consumer getConsumer(Properties kafkaProps) {
+ // on the first connect, and on each reconnect after a record
the consumer reports as failed
+ consumersCreated.incrementAndGet();
+ return super.getConsumer(kafkaProps);
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(0));
+ from(source(in)).routeId(id).process(countAttempt())
+ .doTry()
+ .to(publishUri)
+ .doCatch(Exception.class)
+ .process(exchange -> caught.incrementAndGet())
+ .end();
+
+ from(publishUri).routeId(id + "-publish").to(sink(id + "-out",
"tx") + "&maxRequestSize=1024");
+ }
+ });
+ context.start();
+ contexts.add(context);
+
+ produce(in, poisonRecords());
+ boolean committed3 = awaitQuietly(() ->
Long.valueOf(3).equals(committed(id, in)), 45);
+ List<String> out = readCommittedKeys(id + "-out");
+ LOG.info("Exactly-once rejected redelivery caught by doCatch of the
calling route: committed3={} attempts={} "
+ + "caught={} out={} consumersCreated={} failures={}",
+ committed3, attempts, caught.get(), out,
consumersCreated.get(), sample());
+ assertTrue(committed3, "the poison record is never skipped: attempts="
+ attempts);
+ assertEquals(List.of("small-0", "small-2"), out, "The output of the
poison record must be aborted");
+ assertEquals(1, attempts("big-1"), "The record whose failure the
doCatch handled must not be replayed");
+ assertEquals(1, caught.get());
+ assertEquals(1, consumersCreated.get(),
+ "The skipped record must not be reported as rolled back (which
reconnects the consumer)");
+ }
+
@Test
void kafkaSendInsideDoCatchUsesSourceProducerTransaction() throws
Exception {
String[] names = topics("in", "out");
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 7bb4ec4f9615..6b61780d9d37 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -233,6 +233,16 @@ support it on the path, the component lists the directory
as before. `readLock=c
modification time of a symbolic link itself, as the directory listing reported
them. See the
xref:components::sftp-component.adoc[SFTP component].
+=== camel-kafka - exactlyOnce: a rejected send that a doCatch catches skips
the record
+
+With `exactlyOnce`, a Kafka send through the source producer after an earlier
send failed in the same transaction is
+rejected and marks the exchange rollback-only (a redelivery of the failed
send, or another send such as in a `loop`).
+When a `doCatch` catches that rejected send in the route of the send, the
failure now counts as handled, as when an
+error handler handles it: the output transaction is aborted and the source
offsets are committed alone, so the record
+is skipped. Before, the source transaction was rolled back and the record
replayed. A `doCatch` of a route that called
+the route of the send (for example with `direct`) behaves the same way, as it
did in 4.23. To replay the record, throw
+the exception again in the `doCatch` block, or end it with `rollback()`.
+
=== Java
*Breaking change:* Camel 4.24 requires *Java 21* as the minimum runtime
version.
