davsclaus commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4291166694
##########
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:
Nice catch on the shared counter. One remaining corner, not a blocker: if a
full batch completes while this checkpoint is being stored (the size path sets
the counter to 0) and the count then climbs back to exactly `processed`, the
`compareAndSet` still succeeds and wipes those events from the next window. An
alternative without that case is to take the count when the task runs (`int
processed = processedEvents.getAndSet(0);`) and add it back in the error
callback if the checkpoint fails. Happy to leave it as is if you think the
window is too small to matter.
--
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]
