gnodet-bot commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4292538198
##########
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:
⚠️ **Concurrency note:** `addAndGet(processed)` runs on the reactor thread
without the consumer `lock`, on the same `AtomicInteger` that `processCommit`
increments under `lock`. If the checkpoint store fails *after* a new window has
already started for the same partition (new events have pushed the counter
above zero), the restored count inflates the new window — potentially
triggering a premature size-checkpoint. The result is an extra Blob write, not
a missed checkpoint, so the failure mode is benign. Worth a comment here
explaining the deliberate tradeoff (over-count on store failure → early
checkpoint) to distinguish it from the pre-existing symmetric gap in the
size-checkpoint path (which currently loses the count on failure with no
restore).
--
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]
