oscerd opened a new issue, #1802:
URL: https://github.com/apache/camel-kafka-connector/issues/1802
## Description
`CamelSourceTask.poll()` produces one `CamelSourceRecord` per configured
topic from a single Camel
`Exchange`, and registers **the same exchange** under a separate claim-check
slot for each of them
(`CamelSourceTask.java:239-260`):
```java
for (String singleTopic : topics) {
CamelSourceRecord camelRecord = new CamelSourceRecord(...);
...
Integer claimCheck = freeSlots.remove();
camelRecord.setClaimCheck(claimCheck);
exchangesWaitingForAck[claimCheck] = exchange; // same exchange, N
slots
records.add(camelRecord);
}
```
`commitRecord()` then completes the exchange's unit of work as soon as
**any** of those records is
committed (`CamelSourceTask.java:279`):
```java
UnitOfWorkHelper.doneSynchronizations(correlatedExchange,
correlatedExchange.getExchangeExtension().handoverCompletions());
```
Because `handoverCompletions()` hands the synchronizations over on the first
call, the acknowledgement
towards the external system runs when the first topic's record is committed.
The remaining records
derived from the same exchange are still in flight at that point, and their
later `commitRecord()`
calls have nothing left to complete.
## Expected Behavior
With `topics` listing more than one topic, the source exchange is
acknowledged towards the external
system only after the record has been committed for **every** configured
topic.
## Actual Behavior
The exchange is acknowledged after the first topic's commit. If the worker
fails between that point
and the commit of the remaining topics, the message is gone from the
external system but was never
committed to those topics.
## Steps to Reproduce
1. Configure any source connector with `topics=topicA,topicB`.
2. Use a consumer whose acknowledgement is driven by the exchange's unit of
work (for example a
messaging or queue-based source).
3. Observe that the acknowledgement is issued once, on the first record's
commit, rather than after
both topics have committed.
## Additional Context
Single-topic configurations are unaffected — there is exactly one record per
exchange.
Possible directions: reference-count the exchange across the records derived
from it and complete the
unit of work only when the last one commits, or register the exchange under
a single claim check
shared by all records derived from it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
