oscerd opened a new pull request, #1821:
URL: https://github.com/apache/camel-kafka-connector/pull/1821
Fixes #1803. This one changes offset semantics for sink connectors, so it is
worth reviewing as a
design as much as a diff.
## What was wrong
`put()` sends each record and inspects the exchange, which is sound only
while the route is fully
synchronous. With aggregation configured, the aggregate EIP completes the
**incoming** exchange as
soon as it has been merged into the buffer; the data is delivered later, on
the **aggregated**
exchange. So `put()` returned clean, Kafka Connect committed the offset, and
a subsequent failure of
the aggregated exchange was handled inside the route and dropped — no
exception from `put()`, no
`reporter.report`, no rewind. The whole batch was lost silently.
## How it works now
| Piece | Role |
|---|---|
| `SinkRecordDeliveryTracker` | per-partition set of offsets whose delivery
is outstanding |
| `SinkRecordReference` | identifies a record, and keeps it so a failed
batch can still reach the DLQ |
| `DeliveryTrackingAggregationStrategy` | wraps the strategy named by
`camel.beans.aggregate`, keeping records associated with the aggregated
exchange and releasing them when it completes |
`preCommit` then commits up to, but not including, the oldest record still
awaiting delivery.
**Without aggregation nothing changes:** the route is synchronous, records
are released as `put()`
returns, and `preCommit` passes Connect's own offsets straight through.
`testOffsetsAreNotHeldBackWithoutAggregation` pins that.
## Two things that bit me, in case they matter to review
**The strategy is decorated inside `configure()`, not in `build()`.** The
bean named by
`camel.beans.aggregate` is only bound once the context starts, so a
`lookupByName` in `build()`
returns null and the decorator is silently never applied. What makes this
nasty is that the
hold-back assertion still *passes* in that state — with aggregation on,
nothing releases offsets
anyway — so the feature can look tested while being entirely inert. I found
it only by instrumenting
whether the strategy was actually being called.
**The accumulated records are carried from `oldExchange` on every call.** An
aggregation strategy
commonly does `return newExchange` (as the example in `aggregation.adoc`
does), so the exchange
carrying the batch changes identity as it grows. Storing the list only on
the returned exchange
released **1 of 5** records.
## Tests
`testAggregatedRecordsOffsetsAreHeldBackUntilTheBatchIsDelivered`: four
records with an aggregation
size of five, then `preCommit` must hold at the oldest; a fifth completes
the batch, and once
delivered every offset is committable. Against the unpatched code:
```
offsets must be held at the oldest record whose data is still in the
aggregation buffer
==> expected: <10> but was: <14>
```
— records 10-13 committed while their data sat undelivered in the buffer,
which is the data-loss
window in the issue.
## Worth deciding explicitly
- **Failed batch, no DLQ reporter.** I record the failure and throw from the
next `put()` /
`preCommit`, mirroring the existing no-reporter behaviour, rather than
committing undelivered data.
- **No backpressure.** `preCommit` does not slow `put()`, so a batch that
never completes pins that
partition's offset. That is the correct at-least-once outcome, but
operationally it looks like
stalled progress rather than an error. The default
`camel.aggregation.timeout` of 500ms means
batches do complete in practice.
## Verification
- `core`: 123 tests pass.
- `./mvnw -Psourcecheck -Dcheckstyle.failOnViolation=true
checkstyle:checkstyle`: BUILD SUCCESS.
- Full reactor build 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]
