allthingssecurity commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4291296662
##########
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:
Thanks. You are right that the `compareAndSet` still wipes the next window
when the count climbs back to exactly the checkpointed value after a size
checkpoint during the store round trip. Your `getAndSet(0)` when the task runs,
with the count added back in the error callback, closes that case. I would
rather do that as a small follow-up with its own test (it changes when the
counter is reset, and the failure path needs a test of its own), so this PR
keeps the two fixes it has. If you prefer it here, I can add it.
_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]
