This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/main by this push:
new 0840ed9b4f Fix #1802: acknowledge a source exchange only after every
topic has committed (#1820)
0840ed9b4f is described below
commit 0840ed9b4fe5027b31b4702aa8bd0550c34b11f3
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Aug 25 06:38:02 2026 +0200
Fix #1802: acknowledge a source exchange only after every topic has
committed (#1820)
poll() produces one CamelSourceRecord per configured topic from a single
Exchange, and registered that same exchange under a separate claim check for
each of them. commitRecord() then completed the exchange's unit of work -
which
is what acknowledges the message towards the external system - as soon as
any
one of those records was committed, because handoverCompletions() hands the
synchronizations over on the first call.
With topics listing more than one topic, the message was therefore
acknowledged
after the first topic's commit while the remaining records were still in
flight.
A worker failing in between lost the message: gone from the external system,
never committed to the other topics.
Share one AtomicInteger across the claim checks derived from a single
exchange
and complete the unit of work only when the last of them commits.
Single-topic
configurations are unaffected: the counter starts at one.
getExchangeWaitingForAck is package-private for the test, which needs to
observe
the completion on the exchange the task is actually holding - a
synchronization
registered on the exchange handed to the producer template fires at send
time
and says nothing about when the acknowledgement happens.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../camel/kafkaconnector/CamelSourceTask.java | 29 ++++++++++++-
.../camel/kafkaconnector/CamelSourceTaskTest.java | 48 ++++++++++++++++++++++
2 files changed, 75 insertions(+), 2 deletions(-)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
index 0b489a52ea..876f91fe52 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
@@ -23,6 +23,7 @@ import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
@@ -68,6 +69,9 @@ public class CamelSourceTask extends SourceTask {
private String camelMessageHeaderKey;
private LoggingLevel loggingLevel = LoggingLevel.OFF;
private Exchange[] exchangesWaitingForAck;
+ // One entry per claim check, but all the claim checks derived from the
same exchange share a single counter, so
+ // the exchange's unit of work is only completed once every record
produced from it has been committed.
+ private AtomicInteger[] pendingCommitsWaitingForAck;
//the assumption is that at most 1 thread is running poll() method and at
most 1 thread is running commitRecord()
private SpscArrayQueue<Integer> freeSlots;
private boolean mapProperties;
@@ -139,6 +143,7 @@ public class CamelSourceTask extends SourceTask {
});
//needs to be done like this because freeSlots capacity is rounded
to the next power of 2 of maxNotCommittedRecords
exchangesWaitingForAck = new Exchange[freeSlots.capacity()];
+ pendingCommitsWaitingForAck = new
AtomicInteger[freeSlots.capacity()];
CamelContext camelContext = new DefaultCamelContext();
// componentSchema can legitimately be null in case of kamelet
connectors, in that case KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX + "fromUrl"
property is ignored
@@ -236,6 +241,9 @@ public class CamelSourceTask extends SourceTask {
// reset to be sure that the cache is ready to be used before
sending it in the record (could be useful for SMTs)
sc.reset();
}
+ // every record below is produced from this one exchange; it may
only be acknowledged towards the external
+ // system after the last of them has been committed
+ AtomicInteger pendingCommits = new AtomicInteger(topics.length);
for (String singleTopic : topics) {
CamelSourceRecord camelRecord = new
CamelSourceRecord(sourcePartition, sourceOffset, singleTopic, null,
messageKeySchema,
messageHeaderKey, messageBodySchema, messageBodyValue,
timestamp);
@@ -256,6 +264,7 @@ public class CamelSourceTask extends SourceTask {
Integer claimCheck = freeSlots.remove();
camelRecord.setClaimCheck(claimCheck);
exchangesWaitingForAck[claimCheck] = exchange;
+ pendingCommitsWaitingForAck[claimCheck] = pendingCommits;
LOG.debug("Record: {}, containing data from exchange: {}, is
associated with claim check number: {}", camelRecord, exchange, claimCheck);
records.add(camelRecord);
}
@@ -275,14 +284,23 @@ public class CamelSourceTask extends SourceTask {
Integer claimCheck = ((CamelSourceRecord)record).getClaimCheck();
LOG.debug("Committing record with claim check number: {}", claimCheck);
Exchange correlatedExchange = exchangesWaitingForAck[claimCheck];
+ AtomicInteger pendingCommits = pendingCommitsWaitingForAck[claimCheck];
try {
- UnitOfWorkHelper.doneSynchronizations(correlatedExchange,
correlatedExchange.getExchangeExtension().handoverCompletions());
- LOG.debug("Record with claim check number: {} committed.",
claimCheck);
+ // With more than one configured topic a single exchange produces
one record per topic. Completing the unit
+ // of work acknowledges the message towards the external system,
so it must wait for the last of them.
+ if (pendingCommits == null || pendingCommits.decrementAndGet() <=
0) {
+ UnitOfWorkHelper.doneSynchronizations(correlatedExchange,
correlatedExchange.getExchangeExtension().handoverCompletions());
+ LOG.debug("Record with claim check number: {} committed,
exchange acknowledged.", claimCheck);
+ } else {
+ LOG.debug("Record with claim check number: {} committed, {}
record(s) from the same exchange still pending.",
+ claimCheck, pendingCommits.get());
+ }
} catch (Throwable t) {
LOG.error("Exception during Unit Of Work completion: {} caused by:
{}", t.getMessage(), t.getCause());
throw new RuntimeException(t);
} finally {
exchangesWaitingForAck[claimCheck] = null;
+ pendingCommitsWaitingForAck[claimCheck] = null;
freeSlots.add(claimCheck);
LOG.debug("Claim check number: {} freed.", claimCheck);
}
@@ -383,6 +401,13 @@ public class CamelSourceTask extends SourceTask {
+ "&pollingConsumerBlockWhenFull=" +
pollingConsumerBlockWhenFull;
}
+ /**
+ * The exchange a record's claim check refers to, while it is waiting to
be committed. Visible for testing.
+ */
+ Exchange getExchangeWaitingForAck(int claimCheck) {
+ return exchangesWaitingForAck[claimCheck];
+ }
+
CamelKafkaConnectMain getCms() {
return cms;
}
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
index 70433c4918..c8cace47ba 100644
---
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
@@ -24,12 +24,15 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.kafkaconnector.utils.StringJoinerAggregator;
+import org.apache.camel.support.SynchronizationAdapter;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.header.Header;
@@ -40,8 +43,10 @@ import static org.apache.camel.util.CollectionHelper.mapOf;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class CamelSourceTaskTest {
@@ -664,4 +669,47 @@ public class CamelSourceTaskTest {
sourceTask.stop();
executor.shutdown();
}
+
+ @Test
+ public void testMultiTopicSourceAcknowledgesOnlyAfterTheLastTopicCommits()
{
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME +
",second-topic");
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF,
DIRECT_URI);
+
+ CamelSourceTask sourceTask = new CamelSourceTask();
+ sourceTask.start(props);
+
+ sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, "test");
+
+ List<SourceRecord> poll = sourceTask.poll();
+ assertEquals(2, poll.size(), "one record per configured topic");
+
+ int firstClaimCheck = ((CamelSourceRecord)
poll.get(0)).getClaimCheck();
+ int secondClaimCheck = ((CamelSourceRecord)
poll.get(1)).getClaimCheck();
+ assertNotEquals(firstClaimCheck, secondClaimCheck, "each record gets
its own claim check");
+
+ Exchange held = sourceTask.getExchangeWaitingForAck(firstClaimCheck);
+ assertSame(held, sourceTask.getExchangeWaitingForAck(secondClaimCheck),
+ "both records are derived from the same exchange");
+
+ // completing the unit of work is what acknowledges the message
towards the external system, so observe it on
+ // the exchange the task is actually holding
+ final AtomicInteger acknowledged = new AtomicInteger();
+ held.getExchangeExtension().addOnCompletion(new
SynchronizationAdapter() {
+ @Override
+ public void onDone(Exchange doneExchange) {
+ acknowledged.incrementAndGet();
+ }
+ });
+
+ sourceTask.commitRecord(poll.get(0), null);
+ assertEquals(0, acknowledged.get(),
+ "the exchange must not be acknowledged while a record derived
from it is still uncommitted");
+
+ sourceTask.commitRecord(poll.get(1), null);
+ assertEquals(1, acknowledged.get(),
+ "the exchange must be acknowledged once the last record
derived from it has been committed");
+
+ sourceTask.stop();
+ }
}
