allthingssecurity commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4294519302
##########
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:
Good catch, thanks. Changed the check to `cnt >=
getConfiguration().getCheckpointBatchSize()` in 9fa055837ef6, with a comment on
why. New test `batchSizeCheckpointStillWorksAfterFailedTimeoutCheckpoint` is
your example: batch size 3, the timeout checkpoint of 2 events fails while one
more event is counted, so the count is 3 and the next event takes it to 4.
Without the change that event is not checkpointed by size (fails with "Wanted
but not invoked"); with it, it is checkpointed at once with
`COMPLETED_BY_SIZE`. camel-azure-eventhubs tests: 35 run, 0 failures.
_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]
