allthingssecurity commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4292195477
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -40,14 +40,17 @@ public EventHubsCheckpointUpdaterTask(EventContext
eventContext, AtomicInteger p
@Override
public void run() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a
batch of {}", processedEvents.get());
+ final int processed = processedEvents.get();
+ if (processed > 0) {
+ LOG.debug("checkpointing offset after reaching timeout, with a
batch of {}", processed);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
error -> LOG.warn("Error when updating Checkpoint:
{}", error.getMessage(), error),
() -> {
LOG.debug("Checkpoint updated.");
- processedEvents.set(0);
+ // the counter is shared with the next task of
the partition: an event that completed
+ // while this checkpoint was being stored is
counted for that task, so keep it
+ processedEvents.compareAndSet(processed, 0);
Review Comment:
Update: I changed my mind and did it in this PR, in aaccfdf9fc45, as you
suggested. I measured the window first instead of guessing: a deterministic
simulation (5000 seeded runs, with some slow checkpoint-store answers) lost the
last events of a partition in 31 runs through exactly this case with the
`compareAndSet` version, so it is not too small to matter.
The timer task now takes the count when it starts the checkpoint
(`getAndSet(0)`) and adds it back in the error callback if storing fails. New
test `eventAfterBatchSizeCheckpointDuringTimeoutCheckpointIsCheckpointed`
(batch size 3): while the timeout checkpoint of the first event is stored,
three more events complete, the third after a size checkpoint, and the last
event must still be checkpointed. It fails with the previous commit (`Wanted
but not invoked` after 10 s) and passes now. camel-azure-eventhubs tests: 34
run, 0 failures. The description is updated.
_Claude Code on behalf of allthingssecurity_
--
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]
