oscerd opened a new issue, #1803:
URL: https://github.com/apache/camel-kafka-connector/issues/1803
## Description
`CamelSinkTask.put()` sends each record synchronously and then inspects the
exchange
(`CamelSinkTask.java:192-225`):
```java
producer.send(localEndpoint, exchange);
if (exchange.isFailed()) {
...
throw new ConnectException("Exchange delivery has failed!",
exchange.getException());
}
```
That check is sound only while the route is fully synchronous. When the
operator configures
aggregation (`camel.beans.aggregate` plus `camel.aggregation.size` /
`camel.aggregation.timeout`),
the `ckcAggregator` template is inserted into the route. The aggregate EIP
completes the **incoming**
exchange as soon as it has been merged into the aggregation buffer; delivery
to the configured
endpoint happens later, on the separate aggregated exchange, when
`completionSize` or
`completionTimeout` fires.
So for every record: `producer.send` returns, `isFailed()` is false, `put()`
returns, and Kafka
Connect commits the offset (the default `SinkTask.preCommit` commits
everything handed to `put()`).
If the aggregated exchange subsequently fails at the endpoint, the failure
is handled inside the
route by the error handler and there is no path back to the `SinkTask` — no
exception from `put()`,
no `reporter.report(...)`, no offset rewind.
## Expected Behavior
A sink record's offset is committed only after the record's data has
actually been delivered, or
after it has been explicitly routed to the DLQ.
## Actual Behavior
With aggregation enabled, offsets are committed when records enter the
aggregation buffer. A failure
of the aggregated exchange is logged inside Camel and the whole batch is
dropped without the
connector noticing.
## Additional Context
Possible directions:
- implement `preCommit()` in `CamelSinkTask` to hold back the offsets of
records whose aggregated
exchange has not completed, or
- have the aggregation strategy propagate completion/failure back so `put()`
can block or report.
Note the default error handler is configured with `max.redeliveries=0`, so
by default there is not
even a retry before the batch is dropped.
--
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]
