davsclaus commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4293588144
##########
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:
Small edge case with the add-back: the size check in
`EventHubsConsumer.processCommit` is `cnt ==
getConfiguration().getCheckpointBatchSize()`. Before this change the counter
only went up by one, so it always hit the batch size exactly. Now a failed
store can make it jump past it. For example, with batch size 3, `processed=2`
and one event counted while the store was running: the counter is 1, the
add-back makes it 3, and the next event makes it 4. From then on the `==` never
matches, so the partition is checkpointed only by the timeout until the next
timer run resets the counter. Using `cnt >=
getConfiguration().getCheckpointBatchSize()` there would keep the size
checkpoint working after a failed store.
--
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]
