gnodet-bot commented on code in PR #28206:
URL: https://github.com/apache/camel/pull/28206#discussion_r4333212764
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java:
##########
@@ -46,16 +49,15 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
// we use the EventProcessorClient as recommended by Azure docs to consume
from all partitions
private EventProcessorClient processorClient;
- private final AtomicInteger processedEvents;
+ private final ReentrantLock lock = new ReentrantLock();
Review Comment:
💡 **Note:** This single `ReentrantLock` serializes `processCommit` across
all partitions. Since `processCommit` holds the lock for the full checkpoint
round-trip, a slow partition can delay others. Fine for current use, but worth
a comment if partition density increases.
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -29,26 +31,63 @@ public class EventHubsCheckpointUpdaterTask implements
Runnable {
private static final Logger LOG =
LoggerFactory.getLogger(EventHubsCheckpointUpdaterTask.class);
- private EventContext eventContext;
+ private volatile EventContext eventContext;
private final AtomicInteger processedEvents;
private volatile long scheduledTime;
+ // the lock of the consumer, which counts the events under it
+ private final Lock lock;
+ // set when the task has taken the count of its window, or was cancelled;
read and written under the lock only
+ private boolean done;
+ /**
+ * Creates a task that synchronizes on a private lock of its own,
unrelated to any consumer: for standalone or test
+ * use. {@link EventHubsConsumer} uses {@link
#EventHubsCheckpointUpdaterTask(EventContext, AtomicInteger, Lock)}
+ * with its own lock, which it holds while it counts the events; with a
private lock an event could be counted in a
+ * window whose timer has already taken the count.
+ *
+ * @param eventContext the event context to checkpoint
+ * @param processedEvents the count of the events processed since the last
checkpoint
+ */
public EventHubsCheckpointUpdaterTask(EventContext eventContext,
AtomicInteger processedEvents) {
+ this(eventContext, processedEvents, new ReentrantLock());
+ }
+
+ public EventHubsCheckpointUpdaterTask(EventContext eventContext,
AtomicInteger processedEvents, Lock lock) {
this.eventContext = eventContext;
this.processedEvents = processedEvents;
+ this.lock = lock;
}
@Override
public void run() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a
batch of {}", processedEvents.get());
- eventContext.updateCheckpointAsync()
+ // 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.
+ // This is done under the lock of the consumer, so an event is either
counted (and its context set) before, and
+ // checkpointed here, or after, and then it sees that this task is
done and starts a new window.
+ final int processed;
+ final EventContext context;
+ lock.lock();
+ try {
+ if (done) {
+ LOG.debug("skip checkpointing offset: the window was already
checkpointed by batch size");
+ return;
+ }
+ done = true;
+ processed = processedEvents.getAndSet(0);
+ context = eventContext;
+ } finally {
+ lock.unlock();
+ }
+ if (processed > 0) {
+ LOG.debug("checkpointing offset after reaching timeout, with a
batch of {}", processed);
+ context.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
- error -> LOG.debug("Error when updating
Checkpoint: {}", error.getMessage()),
- () -> {
- 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:
💡 **Note:** `addAndGet(processed)` runs outside the lock (in the outer `try`
after `lock.unlock()` in `finally`). This means it can race with
`processedEvents.set(0)` in `processCommit`. The `>=` comparison at the
batch-size check is what makes this safe — it handles the case where the
counter overshoots before the reset.
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java:
##########
@@ -82,6 +84,10 @@ protected void doStop() throws Exception {
processorClient.stop();
}
+ processedEventsByPartition.clear();
+ scheduledTasksByPartition.clear();
+ checkpointTasksByPartition.clear();
Review Comment:
💡 **Note:** Maps are cleared without calling `cancel()` on the
`ScheduledFuture` values. This relies on the executor shutdown (in
`doShutdown()` via `ShutdownAware`) to cancel pending tasks — which is the
correct Camel lifecycle sequence (`doStop()` then `doShutdown()`). Consider
adding a brief comment to document this dependency.
--
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]
