oscerd opened a new pull request, #1820:
URL: https://github.com/apache/camel-kafka-connector/pull/1820
Fixes #1802.
## What
`poll()` produces one `CamelSourceRecord` per configured topic from a single
`Exchange`, and
registered **that same exchange** under a separate claim check for each:
```java
for (String singleTopic : topics) {
...
Integer claimCheck = freeSlots.remove();
camelRecord.setClaimCheck(claimCheck);
exchangesWaitingForAck[claimCheck] = exchange; // same exchange, N
slots
records.add(camelRecord);
}
```
`commitRecord()` completes that exchange's unit of work — which is what
acknowledges the message
towards the external system — and `handoverCompletions()` hands the
synchronizations over on the
**first** call. So with `topics=a,b` the message was acknowledged after the
first topic's commit,
while the record for the second topic was still in flight. A worker failing
in between loses the
message: gone from the external system, never committed to the other topic.
## Fix
One `AtomicInteger` shared across the claim checks derived from a single
exchange; the unit of work
completes only when the last of them commits. Single-topic configurations
are unaffected — the
counter starts at one, so the first commit is also the last.
## Tests
`testMultiTopicSourceAcknowledgesOnlyAfterTheLastTopicCommits` asserts both
records come from the
same exchange, then registers a `Synchronization` on it and checks the
acknowledgement lands on the
second commit, not the first. Against the unpatched code:
```
the exchange must not be acknowledged while a record derived from it is
still uncommitted
==> expected: <0> but was: <1>
```
**A note on how the test observes this**, since it drove the one
production-code concession here:
attaching the `Synchronization` to the exchange passed to
`ProducerTemplate.send` does not work — it
fires at send time, long before the task holds anything, so such a test
passes with *and* without the
fix. Routing through a `seda:` source moves it later but it still fires
during `poll()`. The probe has
to go on the exchange the task is actually holding, which is why
`getExchangeWaitingForAck` is
package-private rather than the test reaching through reflection. It is a
read-only accessor, and
`getCms()` next to it is package-private for the same reason.
## Verification
- `core`: 122 tests pass.
- `./mvnw -Psourcecheck -Dcheckstyle.failOnViolation=true
checkstyle:checkstyle`: BUILD SUCCESS.
- Full reactor build from the repository root with the whole suite (`./mvnw
clean install`):
BUILD SUCCESS, 28 test runs, no failures.
--
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]
