gnodet-bot commented on code in PR #28207:
URL: https://github.com/apache/camel/pull/28207#discussion_r4333575713
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -32,23 +34,60 @@ public class EventHubsCheckpointUpdaterTask implements
Runnable {
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.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:
⚠️ **Bug: `addAndGet(processed)` without the corresponding `==` → `>=` fix**
This error-recovery path correctly adds `processed` back when a checkpoint
fails. However, the batch-size check in `EventHubsConsumer.processCommit()`
(line 253) still uses:
```java
if (cnt == getConfiguration().getCheckpointBatchSize()) {
```
On `main`, CAMEL-25683 (commit 83aaa80) changed this to:
```java
if (cnt >= getConfiguration().getCheckpointBatchSize()) {
```
with the explicit comment _"a failed timeout checkpoint adds its count back,
so the count can pass the batch size"_. Without the `>=` fix,
`addAndGet(processed)` can push `cnt` past `checkpointBatchSize`, causing the
`==` check to permanently miss and silently disabling batch-size checkpoints.
The same fix needs to be applied here.
##########
docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc:
##########
@@ -1409,6 +1409,19 @@ filtering performed by other components.
Ordinary application headers are unaffected.
+=== camel-azure-eventhubs - an event that completes when the checkpoint timer
of its window has run is checkpointed
Review Comment:
⚠️ **Upgrade note placed under wrong version section**
This `===` subsection is added under `== Upgrading Camel 4.21 to 4.22` (the
general 4.22 section). Since 4.22.2 is already tagged, this fix ships in 4.22.3.
The established convention in this file is per-patch-level sections at the
top (e.g. `== Upgrading from 4.22.0 to 4.22.1` already exists at line 14).
Please add a new `== Upgrading from 4.22.2 to 4.22.3` section at the top of the
file following that pattern.
--
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]
