gnodet-bot commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4294618862
##########
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:
⚠️ **Minor inconsistency:** The timer path
(`EventHubsCheckpointUpdaterTask`) restores the event count on checkpoint
failure via `processedEvents.addAndGet(processed)`, so that those events aren't
silently dropped from the counter. The batch-size path here only logs the error
— if the checkpoint call fails, those `processed` events are lost from the
counter and won't contribute to the next batch-size threshold. Consider adding
the same recovery here:
```suggestion
error -> {
LOG.warn("Error when updating
Checkpoint: {}", error.getMessage(), error);
processedEvents.addAndGet((int)
getConfiguration().getCheckpointBatchSize());
},
```
(Or extract the error-recovery lambda into a shared helper to avoid
duplication.)
--
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]
