Croway commented on code in PR #26028:
URL: https://github.com/apache/camel/pull/26028#discussion_r3912089383
##########
components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java:
##########
@@ -225,30 +290,50 @@ 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
// **************************************
private final class LeadershipListener implements
CamelClusterEventListener.Leadership {
@Override
public void leadershipChanged(CamelClusterView view,
CamelClusterMember leader) {
- if (!isRunAllowed()) {
- return;
- }
-
- if (view.getLocalMember().isLeader()) {
- try {
- onLeadershipTaken();
- } catch (Exception e) {
- getExceptionHandler().handleException("Error starting
consumer while taking leadership", e);
+ lock.lock();
Review Comment:
**Lock-order inversion, can deadlock on stop.**
`leadershipChanged` now takes `BaseService.lock` unconditionally. The view
dispatches this listener while holding its `StampedLock` **read** lock
(`AbstractCamelClusterView.doWithListener` -> `LockHelper.doWithReadLock`). On
the other side, `MasterConsumer.stop()` holds `BaseService.lock` while
`doStop()` calls `view.removeEventListener(...)`, which needs the `StampedLock`
**write** lock.
- Thread S: `stop()` holds `lock` -> `doStop()` -> `removeEventListener` ->
waits for the read lock to be released.
- Thread E (FileLock `tryLock` tick, Curator `LeaderSelector`, Kubernetes
notifier): holds the read lock -> `leadershipChanged` -> `lock.lock()` -> waits
for S.
Both threads block forever, shutdown hangs, and every other master route on
that view stops receiving events. Pre-PR the unlocked `isRunAllowed()`
pre-check returned before contending for the lock once `STOPPING` was set (and
lost-events never locked without a consumer), so the window was a microsecond
race. Now every event contends, and combined with the start running under the
lock (see below) the event thread can already be parked on `lock` for the whole
delegated start when `stop()` arrives.
Suggestion: keep an unlocked `isRunAllowed()` fast-path before taking the
lock, and/or use `tryLock` with a bail-out here, and make sure nothing that
runs under `BaseService.lock` needs the view's write lock (e.g. remove the
listener before/outside the locked section of `doStop`).
##########
components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java:
##########
@@ -151,60 +162,105 @@ private BackgroundTask createTask() {
.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) {
+ try {
+ // interruptibly, so cancelling the task while this consumer is
being stopped does not
+ // keep the leader pool thread waiting for a lock the stopping
thread holds
+ lock.lockInterruptibly();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return true; // no more attempts
+ }
+ 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
+ }
- } catch (Exception e) {
- cause = e;
+ if (delegatedConsumer != null) {
+ return true; // no more attempts
+ }
+
+ LOG.info("Leadership taken. Attempt #{} to start consumer: {}",
task.iteration(), delegatedEndpoint);
+
+ Exception cause = null;
+ try {
+ Consumer consumer = attempt.get();
+ if (consumer == null) {
+ consumer = delegatedEndpoint.createConsumer(processor);
+ 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);
Review Comment:
**The delegated start now runs while holding `BaseService.lock`.**
Before this PR the start lambda ran on the leader pool without the lock;
only `schedule()` was under it. Now `createConsumer()` and
`ServiceHelper.startService(delegatedEndpoint, consumer)` run with the lock
held. A delegated JMS/Kafka consumer that spends 30 s+ in connect retries
inside `doStart()` means:
- `MasterConsumer.stop()` (context shutdown) blocks on `lock` until the
start returns, so the `cancelLeaderTask(true)` in `doStop` can never interrupt
an in-flight start (`lockInterruptibly` only helps a task that has not acquired
the lock yet).
- The view's dispatch thread parks on `lock` under the `StampedLock` read
lock, so `addEventListener`/`removeEventListener` of every other
`MasterConsumer` on the namespace block.
- For `FileLockClusterView` the same `tryLock()` thread that fires the event
also writes the leader heartbeat (`writeClusterLeaderInfo`), so other members
can judge this leader stale and take over while this node is still starting.
Suggestion: claim the start under the lock (check `leadershipTaken`,
`delegatedConsumer == null`, set a "starting" marker), run create/start **off**
the lock, then re-acquire to publish `delegatedConsumer`, and if leadership was
lost in the meantime stop/shutdown the freshly started consumer instead of
publishing it.
##########
components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java:
##########
@@ -151,60 +162,105 @@ private BackgroundTask createTask() {
.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) {
+ try {
+ // interruptibly, so cancelling the task while this consumer is
being stopped does not
+ // keep the leader pool thread waiting for a lock the stopping
thread holds
+ lock.lockInterruptibly();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return true; // no more attempts
+ }
+ 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
+ }
- } catch (Exception e) {
- cause = e;
+ if (delegatedConsumer != null) {
+ return true; // no more attempts
+ }
+
+ LOG.info("Leadership taken. Attempt #{} to start consumer: {}",
task.iteration(), delegatedEndpoint);
+
+ Exception cause = null;
+ try {
+ Consumer consumer = attempt.get();
+ if (consumer == null) {
+ consumer = delegatedEndpoint.createConsumer(processor);
+ 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);
+ // publish the consumer only once it is started, so a failed
attempt cannot leave an unstarted
+ // consumer behind that makes every later leadership term
believe there is nothing left to do
+ delegatedConsumer = 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);
+ if (cause != null) {
+ String message = "Leadership taken. Attempt #" +
task.iteration()
+ + " failed to start consumer due to: " +
cause.getMessage();
+ getExceptionHandler().handleException(message, cause);
+ if (task.iteration() >=
masterEndpoint.getComponent().getBackOffMaxAttempts()) {
Review Comment:
**This log never fires with default settings, `backOffMaxAttempts` is
effectively ignored.**
`createTask()` uses `Budgets.iterationTimeBudget()` without
`withMaxDuration`/`withUnlimitedDuration`, so the builder's
`DEFAULT_MAX_DURATION = 5000` ms applies, measured from task construction. With
the defaults (`backOffDelay=5000`, `backOffMaxAttempts=10`, initial delay 1 s):
attempt #1 runs at t~1 s and fails; the tick at t~6 s finds the budget
exhausted and the task ends. `task.iteration()` is 1, never >= 10, so this
ERROR is never emitted and the node silently sits as an idle leader (only
`BackgroundTask`'s generic WARN appears). The budget config itself is
pre-existing, but the PR's give-up contract and this message rely on it, and
`MasterConsumerLeadershipTest` masks it with `backOffDelay=200` /
`maxAttempts=2`, which fits inside 5 s.
Also, `withMaxIterations()` ignores values <= 0, so `backOffMaxAttempts=0`
keeps unlimited iterations but `iteration() >= 0` is always true here: "Giving
up after 1 attempts" is logged on every failed attempt while retries continue.
Suggestion: add `.withUnlimitedDuration()` (or a duration derived from
`backOffDelay * backOffMaxAttempts`) in `createTask()`, and gate this log on
the task actually being exhausted (its status/budget) rather than re-deriving
the arithmetic here. A test with the default component settings would catch
this.
--
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]
