Skip to content

CAMEL-24583: camel-master - do not start the delegated consumer after leadership is lost (CAMEL-24584) #26052

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
davsclaus merged 1 commit into from
Sep 3, 2026
Merged

CAMEL-24583: camel-master - do not start the delegated consumer after leadership is lost (CAMEL-24584) #26052

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
212 changes: 162 additions & 50 deletions components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,9 @@
package org.apache.camel.component.master;

import java.time.Duration;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
Expand Down Expand Up @@ -59,6 +61,10 @@ public class MasterConsumer extends DefaultConsumer implements ResumeAware<Resum
private volatile CamelClusterView view;
private ResumeStrategy resumeStrategy;
private ScheduledExecutorService leaderPool;
// leadership state and the pending start task are guarded by lock, which is also held by the
// service lifecycle methods, so a leadership event cannot interleave with start/stop of this consumer
private boolean leadershipTaken;
private Future<?> leaderTaskFuture;

public MasterConsumer(MasterEndpoint masterEndpoint, Processor processor, CamelClusterService clusterService) {
super(masterEndpoint, processor);
Expand Down Expand Up @@ -103,6 +109,15 @@ protected void doStart() throws Exception {
protected void doStop() throws Exception {
super.doStop();

// a start can still be pending, cancel it first so it cannot start the delegated consumer
// after this consumer has been stopped
leadershipTaken = false;
cancelLeaderTask(true);

// note: removeEventListener below needs the cluster view lock while this thread holds the lock of
// this service, which is the opposite order of an event dispatch. Nothing that runs under this lock
// may wait for the view, and the listener bails out before locking once this consumer is stopping

if (view != null) {
view.removeEventListener(leadershipListener);
clusterService.releaseView(view);
Expand Down Expand Up @@ -146,65 +161,126 @@ private BackgroundTask createTask() {
.withInterval(Duration.ofMillis(masterEndpoint.getComponent().getBackOffDelay()))
.withInitialDelay(Duration.ofSeconds(1))
.withMaxIterations(masterEndpoint.getComponent().getBackOffMaxAttempts())
// the attempts are bounded by backOffMaxAttempts, not by the 5s default duration of
// the builder, which would otherwise end the task before the second attempt
.withUnlimitedDuration()
.build())
.withName("Leadership")
.build();
}

private void onLeadershipTaken() throws Exception {
private void onLeadershipTaken() {
lock.lock();
try {
if (!isRunAllowed()) {
return;
}

if (delegatedConsumer != null) {
leadershipTaken = true;

if (delegatedConsumer != null || isStartPending()) {
return;
}

final BackgroundTask leaderTask = createTask();
leaderTask.schedule(getEndpoint().getCamelContext(), () -> {
if (!isRunAllowed()) {
return false;
}
LOG.info("Leadership taken. Attempt #{} to start consumer: {}", leaderTask.iteration(), delegatedEndpoint);

Exception cause = null;
try {
if (delegatedConsumer == null) {
delegatedConsumer = delegatedEndpoint.createConsumer(processor);
if (delegatedConsumer instanceof StartupListener) {
getEndpoint().getCamelContext().addStartupListener((StartupListener) delegatedConsumer);
}
if (delegatedConsumer instanceof ResumeAware resumeAwareConsumer && resumeStrategy != null) {
LOG.debug("Setting up the resume adapter for the resume strategy in consumer");
ResumeAdapter resumeAdapter
= AdapterHelper.eval(clusterService.getCamelContext(), resumeAwareConsumer,
resumeStrategy);
resumeStrategy.setAdapter(resumeAdapter);

LOG.debug("Setting up the resume strategy for consumer");
resumeAwareConsumer.setResumeStrategy(resumeStrategy);
}
}
ServiceHelper.startService(delegatedEndpoint, delegatedConsumer);
// a task from a previous leadership term may still be scheduled, drop it
cancelLeaderTask(false);

final BackgroundTask task = createTask();
// the consumer is created once and re-used by the start attempts of this task
final AtomicReference<Consumer> attempt = new AtomicReference<>();
leaderTaskFuture = task.schedule(getEndpoint().getCamelContext(), () -> startDelegatedConsumer(task, attempt));
} finally {
lock.unlock();
}
}

private boolean startDelegatedConsumer(BackgroundTask task, AtomicReference<Consumer> attempt) {
lock.lock();
try {
if (!isRunAllowed()) {
return false;
}

if (!leadershipTaken) {
// leadership was lost while this start was pending. Starting now would run the consumer on a
// node that is not the leader, and no further leadership event is coming to stop it again
LOG.debug("Leadership lost while the start was pending. Not starting consumer: {}", delegatedEndpoint);
return true; // no more attempts
}

if (delegatedConsumer != null) {
return true; // no more attempts
}
} finally {
lock.unlock();
}

LOG.info("Leadership taken. Attempt #{} to start consumer: {}", task.iteration(), delegatedEndpoint);

} catch (Exception e) {
cause = e;
// the delegate is created and started without holding the lock. It can block for a long time, and the
// lock is taken by the service lifecycle and by the cluster view event dispatch, which must not wait
// for a broker connect. The leadership is re-checked below before the consumer is published
Consumer consumer = attempt.get();
Exception cause = null;
try {
if (consumer == null) {
consumer = delegatedEndpoint.createConsumer(processor);
// held for the attempts of this task, so the startup listener and the resume strategy are
// wired once and a retry only starts the consumer again
attempt.set(consumer);
if (consumer instanceof StartupListener startupListener) {
getEndpoint().getCamelContext().addStartupListener(startupListener);
}
if (consumer instanceof ResumeAware resumeAwareConsumer && resumeStrategy != null) {
LOG.debug("Setting up the resume adapter for the resume strategy in consumer");
ResumeAdapter resumeAdapter
= AdapterHelper.eval(clusterService.getCamelContext(), resumeAwareConsumer,
resumeStrategy);
resumeStrategy.setAdapter(resumeAdapter);

LOG.debug("Setting up the resume strategy for consumer");
resumeAwareConsumer.setResumeStrategy(resumeStrategy);
}
}
ServiceHelper.startService(delegatedEndpoint, consumer);
} catch (Exception e) {
cause = e;
}

if (cause != null) {
String message = "Leadership taken. Attempt #" + leaderTask.iteration()
+ " failed to start consumer due to: " + cause.getMessage();
getExceptionHandler().handleException(message, cause);
// make the task runner aware of the exception (will retry)
throw new TaskRunFailureException(message, cause);
lock.lock();
try {
if (cause != null) {
// the consumer is kept for the next attempt. It is not stopped here: a consumer that failed to
// start was already stopped by its own start(), and shutting it down would also shut down the
// processor of the route, which the next attempt and this consumer still need
String message = "Leadership taken. Attempt #" + task.iteration()
+ " failed to start consumer due to: " + cause.getMessage();
getExceptionHandler().handleException(message, cause);
int maxAttempts = masterEndpoint.getComponent().getBackOffMaxAttempts();
if (maxAttempts > 0 && task.iteration() >= maxAttempts) {
LOG.error("Leadership taken. Giving up after {} attempts to start consumer: {}."
+ " This node holds the leadership but is not consuming, until the leadership changes again.",
task.iteration(), delegatedEndpoint);
}
// make the task runner aware of the exception (will retry)
throw new TaskRunFailureException(message, cause);
}

LOG.info("Leadership taken. Attempt #{} success. Consumer started: {}", leaderTask.iteration(),
delegatedEndpoint);
if (!leadershipTaken || !isRunAllowed()) {
// the leadership went away while the consumer was starting, so stop what was just started
// instead of publishing it. No leadership event is going to do it, delegatedConsumer is unset
LOG.info("Leadership lost while the consumer was starting. Stopping consumer: {}", delegatedEndpoint);
ServiceHelper.stopAndShutdownServices(consumer, delegatedEndpoint);
attempt.set(null);
return true; // no more attempts
});
}

delegatedConsumer = consumer;
LOG.info("Leadership taken. Attempt #{} success. Consumer started: {}", task.iteration(),
delegatedEndpoint);
// release the task, a later leadership term schedules a new one
cancelLeaderTask(false);
return true; // no more attempts
} finally {
lock.unlock();
}
Expand All @@ -213,6 +289,15 @@ private void onLeadershipTaken() throws Exception {
private void onLeadershipLost() {
lock.lock();
try {
leadershipTaken = false;
// a start scheduled by the leadership taken event may not have run yet, cancel it so it
// cannot start the consumer on a node that is no longer the leader
cancelLeaderTask(false);

if (delegatedConsumer == null) {
return;
}

LOG.debug("Leadership lost. Stopping consumer: {}", delegatedEndpoint);
try {
ServiceHelper.stopAndShutdownServices(delegatedConsumer, delegatedEndpoint);
Expand All @@ -225,6 +310,17 @@ private void onLeadershipLost() {
}
}

private boolean isStartPending() {
return leaderTaskFuture != null && !leaderTaskFuture.isDone();
}

private void cancelLeaderTask(boolean mayInterruptIfRunning) {
if (leaderTaskFuture != null) {
leaderTaskFuture.cancel(mayInterruptIfRunning);
leaderTaskFuture = null;
}
}

// **************************************
// Listener
// **************************************
Expand All @@ -233,22 +329,38 @@ private final class LeadershipListener implements CamelClusterEventListener.Lead
@Override
public void leadershipChanged(CamelClusterView view, CamelClusterMember leader) {
if (!isRunAllowed()) {
// checked before taking the lock: this runs on the cluster view dispatch thread while that
// view holds its own lock, and a consumer that is stopping holds this lock while it removes
// this listener from the view
return;
}

if (view.getLocalMember().isLeader()) {
try {
onLeadershipTaken();
} catch (Exception e) {
getExceptionHandler().handleException("Error starting consumer while taking leadership", e);
lock.lock();
try {
if (!isRunAllowed()) {
return;
}
} else if (delegatedConsumer != null) {
try {
onLeadershipLost();
} catch (Exception e) {
getExceptionHandler()
.handleException("Error stopping consumer while loosing leadership. This exception is ignored.", e);

// the leadership is read under the same lock that applies it, so that two events
// dispatched concurrently cannot be applied in the wrong order
if (view.getLocalMember().isLeader()) {
try {
onLeadershipTaken();
} catch (Exception e) {
getExceptionHandler().handleException("Error starting consumer while taking leadership", e);
}
} else {
// dispatched even when there is no consumer yet, as a start may be pending
try {
onLeadershipLost();
} catch (Exception e) {
getExceptionHandler()
.handleException("Error stopping consumer while loosing leadership. This exception is ignored.",
e);
}
}
} finally {
lock.unlock();
}
}
}
Expand Down
Loading