allthingssecurity commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4292682331


##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -40,15 +40,19 @@ 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());
+        // take the count when the checkpoint is started: an event that 
completes while it is being stored is counted
+        // for the next window from 0, and a checkpoint by batch size in the 
meantime cannot be confused with this one
+        final int processed = processedEvents.getAndSet(0);
+        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);
-                            });
+                            error -> {
+                                LOG.warn("Error when updating Checkpoint: {}", 
error.getMessage(), error);
+                                // these events are not checkpointed: count 
them again for the next checkpoint
+                                processedEvents.addAndGet(processed);

Review Comment:
   Agreed, that is the intended trade-off: after a failed store the restored 
count can only make the next size checkpoint come a little earlier (one extra 
checkpoint write), never skip one. The code comment at that line says the 
events are counted again for the next checkpoint; I can extend it with the "may 
over-count, never under-count" wording if a maintainer wants 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]

Reply via email to