Skip to content

[backport camel-4.18.x] CAMEL-25781: camel-azure-eventhubs - an event that completes when the checkpoint timer of its window has run is checkpointed by the next timer #28206

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gnodet merged 3 commits into from
Oct 21, 2026
Merged

[backport camel-4.18.x] CAMEL-25781: camel-azure-eventhubs - an event that completes when the checkpoint timer of its window has run is checkpointed by the next timer #28206

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6f05afe
[backport camel-4.18.x] CAMEL-25781: camel-azure-eventhubs - an event…
gnodet Oct 21, 2026
ca67e79
Address review: add missing tests, fix upgrade guide section placement
gnodet Oct 21, 2026
9c1f25a
Address review: fix upgrade guide logging wording; remove redundant a…
gnodet Oct 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions components/camel-azure/camel-azure-eventhubs/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,5 +72,10 @@
<artifactId>commons-lang3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>
77 changes: 68 additions & 9 deletions .../main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@
package org.apache.camel.component.azure.eventhubs;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.azure.messaging.eventhubs.models.EventContext;
import org.slf4j.Logger;
Expand All @@ -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);
Comment thread
gnodet marked this conversation as resolved.
},
() -> LOG.debug("Checkpoint updated."));
} else {
LOG.debug("skip checkpointing offset even if timeout is reached. No events processed");
}
Expand All @@ -66,6 +105,26 @@ public long getScheduledTime() {
return scheduledTime;
}

/**
* Whether this task has taken the count of its window (or was cancelled), so a new event must start a new window.
* Called under the lock of the consumer.
*/
public boolean isDone() {
return done;
}

/**
* Cancels the task under the lock of the consumer: if it is already running, it takes nothing.
*/
public void cancel() {
lock.lock();
try {
done = true;
} finally {
lock.unlock();
}
}

public boolean isExpired() {
return System.currentTimeMillis() > scheduledTime;
}
Expand Down
113 changes: 72 additions & 41 deletions ...eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,13 @@
*/
package org.apache.camel.component.azure.eventhubs;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import com.azure.messaging.eventhubs.EventProcessorClient;
import com.azure.messaging.eventhubs.models.ErrorContext;
Expand All @@ -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();
Comment thread
gnodet marked this conversation as resolved.
private final AtomicInteger pendingExchanges = new AtomicInteger();
private final Map<String, AtomicInteger> processedEventsByPartition = new ConcurrentHashMap<>();
private final Map<String, ScheduledFuture<?>> scheduledTasksByPartition = new ConcurrentHashMap<>();
private final Map<String, EventHubsCheckpointUpdaterTask> checkpointTasksByPartition = new ConcurrentHashMap<>();
private ScheduledExecutorService scheduledExecutorService;
private ScheduledFuture<?> lastScheduledTask;
private EventHubsCheckpointUpdaterTask lastTask;

public EventHubsConsumer(final EventHubsEndpoint endpoint, final Processor processor) {
super(endpoint, processor);

this.processedEvents = new AtomicInteger();
}

@Override
Expand All @@ -82,6 +84,10 @@ protected void doStop() throws Exception {
processorClient.stop();
}

processedEventsByPartition.clear();
scheduledTasksByPartition.clear();
checkpointTasksByPartition.clear();
Comment thread
gnodet marked this conversation as resolved.

// shutdown camel consumer
super.doStop();
}
Expand Down Expand Up @@ -214,46 +220,71 @@ private void onErrorListener(final ErrorContext errorContext) {
/**
* Strategy to commit the offset after message being processed successfully.
*
* @param exchange the exchange
* @param exchange the exchange
* @param eventContext the event context of the event, used to update the checkpoint
*/
private void processCommit(final Exchange exchange, final EventContext eventContext) {
if (lastTask == null || lastTask.isExpired()) {
lastTask = new EventHubsCheckpointUpdaterTask(eventContext, processedEvents);
// delegate the checkpoint update to a dedicated Thread
long timeout = getConfiguration().getCheckpointBatchTimeout();
lastTask.setScheduledTime(System.currentTimeMillis() + timeout);
lastScheduledTask = scheduledExecutorService.schedule(lastTask, timeout, TimeUnit.MILLISECONDS);
} else {
// updates the eventContext to use for the offset to be the most accurate
lastTask.setEventContext(eventContext);
}

void processCommit(final Exchange exchange, final EventContext eventContext) {
lock.lock();
try {
var cnt = processedEvents.incrementAndGet();
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");
if (lastScheduledTask != null) {
lastScheduledTask.cancel(false);
}
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
error -> LOG.debug("Error when updating Checkpoint: {}", error.getMessage()),
() -> {
LOG.debug("Checkpoint updated.");
});
} else if (lastTask != null && lastTask.isExpired()) {
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY, COMPLETED_BY_TIMEOUT);
LOG.debug("eventhub consumer batch timeout reached");
final String partitionId = eventContext.getPartitionContext().getPartitionId();
final AtomicInteger processedEvents = processedEventsByPartition.computeIfAbsent(partitionId,
ignored -> new AtomicInteger());
EventHubsCheckpointUpdaterTask checkpointTask = checkpointTasksByPartition.get(partitionId);
ScheduledFuture<?> scheduledTask = scheduledTasksByPartition.get(partitionId);

// a new window starts when the timer of the current one has run, not when the wall clock passed its time: the
// timer can run in the millisecond of its scheduled time, or later, and the event must be counted in the
// window whose timer has not taken the count yet
if (checkpointTask == null || checkpointTask.isDone()) {
checkpointTask = new EventHubsCheckpointUpdaterTask(eventContext, processedEvents, lock);
// delegate the checkpoint update to a dedicated Thread
long timeout = getConfiguration().getCheckpointBatchTimeout();
checkpointTask.setScheduledTime(System.currentTimeMillis() + timeout);
scheduledTask = scheduledExecutorService.schedule(checkpointTask, timeout, TimeUnit.MILLISECONDS);
checkpointTasksByPartition.put(partitionId, checkpointTask);
scheduledTasksByPartition.put(partitionId, scheduledTask);
} else {
LOG.debug("neither eventhub consumer batch size of {}/{} nor batch timeout reached yet", cnt,
getConfiguration().getCheckpointBatchSize());
// updates the eventContext to use for the offset to be the most accurate
checkpointTask.setEventContext(eventContext);
}

try {
var cnt = processedEvents.incrementAndGet();
// >= 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);
// a timer that has already started takes nothing (it waits for the lock)
checkpointTask.cancel();
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),
() -> {
LOG.debug("Checkpoint updated.");
});
} else if (checkpointTask.isExpired()) {
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY, COMPLETED_BY_TIMEOUT);
LOG.debug("eventhub consumer batch timeout reached for partition {}", partitionId);
} else {
LOG.debug("neither eventhub consumer batch size of {}/{} nor batch timeout reached yet for partition {}",
cnt, getConfiguration().getCheckpointBatchSize(), partitionId);
}
// we assume that the scheduled task has done the update by its side
} catch (Exception ex) {
getExceptionHandler().handleException(
"Error occurred during updating the checkpoint. This exception is ignored.",
exchange, ex);
}
// we assume that the scheduled task has done the update by its side
} catch (Exception ex) {
getExceptionHandler().handleException("Error occurred during updating the checkpoint. This exception is ignored.",
exchange, ex);
} finally {
lock.unlock();
}
}

Expand Down
Loading
Loading