allthingssecurity commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4294662591
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java:
##########
@@ -246,13 +246,18 @@ private void processCommit(final Exchange exchange, final
EventContext eventCont
try {
var cnt = processedEvents.incrementAndGet();
- if (cnt == getConfiguration().getCheckpointBatchSize()) {
+ // >= and not ==: a failed timeout checkpoint adds its count
back, so the count can pass the batch size
+ if (cnt >= getConfiguration().getCheckpointBatchSize()) {
processedEvents.set(0);
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY,
COMPLETED_BY_SIZE);
LOG.debug("eventhub consumer batch size of reached for
partition {}", partitionId);
if (scheduledTask != null) {
scheduledTask.cancel(false);
}
+ // the next event starts a new timeout window: the
cancelled task must not be reused until its
+ // scheduled time, else the events of the rest of that
window are not checkpointed by the timeout
+ checkpointTasksByPartition.remove(partitionId);
+ scheduledTasksByPartition.remove(partitionId);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one
event..."),
error -> LOG.warn("Error when updating
Checkpoint: {}", error.getMessage(), error),
Review Comment:
Thanks. I'd keep the batch-size path as it is in this PR, because adding the
count back there would not change what gets checkpointed:
- A checkpoint stores the offset of the latest event, so the next checkpoint
of the partition (by size or by timeout) also covers the events of a failed
one. The count only decides *when* that next checkpoint happens; restoring it
would just make the next size checkpoint come a little earlier.
- The case that is really left open is a failed size checkpoint followed by
no further event: the timer task of that window was cancelled, so nothing
checkpoints those events until the next event arrives (on a restart they are
delivered again). Restoring the counter does not help there either, since no
task is scheduled to look at it. That is the same on main today (the batch-size
path only logs), so I've noted it as a separate follow-up rather than widening
this PR: a failed size checkpoint should schedule a timeout checkpoint for the
same events.
In the timer path the add-back matters because that task skips the
checkpoint when its count is 0.
_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]
