[
https://issues.apache.org/jira/browse/CAMEL-24583?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Henrik updated CAMEL-24583:
---------------------------
Description:
{{MasterConsumer}} starts the delegated consumer asynchronously, from a
{{BackgroundTask}} scheduled one second after the leadership-taken event. A
leadership-lost event that arrives inside that window is discarded, and the
task then starts the consumer on a node that is no longer the leader. Nothing
stops it afterwards, because no further leadership event is coming: the node is
already a follower.
The result is two active consumers on an endpoint that {{master:}} exists to
keep single-consumer.
Line numbers below are 4.22.0.
# {{onLeadershipTaken()}} (line 154) holds {{lock}} only long enough to
schedule the task (line 166). {{delegatedConsumer}} is still null when it
returns.
# The scheduled body's only guard is {{if (!isRunAllowed()) return false;}}
(line 167). {{isRunAllowed()}} is the consumer's own {{BaseService}} status,
that is, whether the route is running. It says nothing about leadership.
# {{LeadershipListener.leadershipChanged()}} (line 235) dispatches to
{{onLeadershipLost()}} only {{else if (delegatedConsumer != null)}} (line 245).
While the start is pending that field is null, so the lost event is dropped and
nothing cancels the pending task.
# One second later (line 147, {{{}withInitialDelay(Duration.ofSeconds(1)){}}},
not configurable) the task creates and starts the delegated consumer. The node
is a follower.
h2. Impact
Observed in production on Camel 4.21.0 with {{KubernetesClusterService}} and a
google-pubsub delegate. A three second leadership flap produced repeated
taken/lost cycles on the same endpoint, at sub-second intervals well inside
the one second start delay:
{code}
16:01:27.567 Leadership lost. Consumer stopped:
google-pubsub://<project>:<Queue>
16:01:28.671 Leadership taken. Attempt #1 to start consumer:
google-pubsub://<project>:<Queue>
16:01:29.932 Stopping subscribers for <project>/<Queue>
{code}
That is the precondition for the race: a lost event arriving while a start is
pending is dropped, because the listener guards it on delegatedConsumer != null.
h2. Reproducer
Fake the cluster view so leadership can be flapped on demand, then assert that
the delegated consumer is never started:
{code:java}
// view.getLocalMember().isLeader() returns leader.get()
// masterEndpoint.getEndpoint() returns a mock Endpoint whose createConsumer()
is observable
consumer = new MasterConsumer(masterEndpoint, processor, clusterService);
consumer.start();
leader.set(true);
leadershipListener.leadershipChanged(view, localMember); // taken
leader.set(false);
leadershipListener.leadershipChanged(view, localMember); // lost, start still
pending
Thread.sleep(3000); // outlast the 1s
initial delay
verify(delegatedEndpoint, never()).createConsumer(any()); // fails: the
consumer was started
{code}
h2. Two related observations in the same class
* {{BackgroundTask.schedule()}} returns the {{scheduleWithFixedDelay}} future,
but {{MasterConsumer }}neither retains nor cancels it. The task only
short-circuits through its own latch ({{{}BackgroundTask.runTaskWrapper{}}},
{{{}if (latch.getCount() == 0) return;{}}}), so every leadership change
leaves a repeating no-op task on the "Leadership" pool for the life of the
route.
* If {{ServiceHelper.startService()}} throws, {{delegatedConsumer}} has
already been assigned and is left non-null but unstarted. Once the retry budget
is exhausted, {{{}onLeadershipTaken(){}}}'s {{if (delegatedConsumer != null)
return;}} early-return means a later leadership-taken event schedules
nothing, and the node stays nominally leader while consuming nothing until it
restarts.
h2. Proposed fix
* Track the leadership taken/lost state, guarded by the same lock the start
and the stop take, and re-check it inside the scheduled task before creating
the delegated consumer.
* Read the leadership inside that lock in the listener, and dispatch the lost
event unconditionally, so a lost event during a pending start cancels it
instead of being dropped.
* Retain the {{Future}} from {{BackgroundTask.schedule()}} and cancel it on
leadership lost, on a successful start, and on stop.
* Publish {{delegatedConsumer}} only after it has started, so a failed attempt
cannot make every later leadership term a no-op.
A patch with tests is ready; the {{BackgroundTask}} half (a scheduled task
should unschedule itself once it has completed or run out of budget) is
proposed as a separate camel-support issue.
was:
{{MasterConsumer}} starts the delegated consumer asynchronously, from a
{{BackgroundTask}} scheduled one second after the leadership-taken event. A
leadership-lost event that arrives inside that window is discarded, and the
task then starts the consumer on a node that is no longer the leader. Nothing
stops it afterwards, because no further leadership event is coming: the node is
already a follower.
The result is two active consumers on an endpoint that {{master:}} exists to
keep single-consumer.
Line numbers below are 4.22.0.
# {{onLeadershipTaken()}} (line 154) holds {{lock}} only long enough to
schedule the task (line 166). {{delegatedConsumer}} is still null when it
returns.
# The scheduled body's only guard is {{if (!isRunAllowed()) return false;}}
(line 167). {{isRunAllowed()}} is the consumer's own {{BaseService}} status,
that is, whether the route is running. It says nothing about leadership.
# {{LeadershipListener.leadershipChanged()}} (line 235) dispatches to
{{onLeadershipLost()}} only {{else if (delegatedConsumer != null)}} (line 245).
While the start is pending that field is null, so the lost event is dropped and
nothing cancels the pending task.
# One second later (line 147, {{{}withInitialDelay(Duration.ofSeconds(1)){}}},
not configurable) the task creates and starts the delegated consumer. The node
is a follower.
h2. Impact
Seen in production with a Google Pub/Sub delegate behind
{{{}KubernetesClusterService{}}}. A three second membership flap left
subscribers running on the follower pod. Those subscribers kept pulling
messages into routes whose error handlers had been stopped, so every exchange
failed with {{java.util.concurrent.RejectedExecutionException}} from
{{RedeliveryErrorHandler$RedeliveryTask.run}} (the {{isRunAllowed()}} branch)
and the message was lost rather than redelivered to the leader. The pod
consumed and discarded messages for 54 minutes until it was restarted.
h2. Reproducer
Fake the cluster view so leadership can be flapped on demand, then assert that
the delegated consumer is never started:
{code:java}
// view.getLocalMember().isLeader() returns leader.get()
// masterEndpoint.getEndpoint() returns a mock Endpoint whose createConsumer()
is observable
consumer = new MasterConsumer(masterEndpoint, processor, clusterService);
consumer.start();
leader.set(true);
leadershipListener.leadershipChanged(view, localMember); // taken
leader.set(false);
leadershipListener.leadershipChanged(view, localMember); // lost, start still
pending
Thread.sleep(3000); // outlast the 1s
initial delay
verify(delegatedEndpoint, never()).createConsumer(any()); // fails: the
consumer was started
{code}
h2. Two related observations in the same class
* {{BackgroundTask.schedule()}} returns the {{scheduleWithFixedDelay}} future,
but {{MasterConsumer }}neither retains nor cancels it. The task only
short-circuits through its own latch ({{{}BackgroundTask.runTaskWrapper{}}},
{{{}if (latch.getCount() == 0) return;{}}}), so every leadership change
leaves a repeating no-op task on the "Leadership" pool for the life of the
route.
* If {{ServiceHelper.startService()}} throws, {{delegatedConsumer}} has
already been assigned and is left non-null but unstarted. Once the retry budget
is exhausted, {{{}onLeadershipTaken(){}}}'s {{if (delegatedConsumer != null)
return;}} early-return means a later leadership-taken event schedules
nothing, and the node stays nominally leader while consuming nothing until it
restarts.
h2. Proposed fix
* Track the leadership taken/lost state, guarded by the same lock the start
and the stop take, and re-check it inside the scheduled task before creating
the delegated consumer.
* Read the leadership inside that lock in the listener, and dispatch the lost
event unconditionally, so a lost event during a pending start cancels it
instead of being dropped.
* Retain the {{Future}} from {{BackgroundTask.schedule()}} and cancel it on
leadership lost, on a successful start, and on stop.
* Publish {{delegatedConsumer}} only after it has started, so a failed attempt
cannot make every later leadership term a no-op.
A patch with tests is ready; the {{BackgroundTask}} half (a scheduled task
should unschedule itself once it has completed or run out of budget) is
proposed as a separate camel-support issue.
> camel-master: losing leadership while a consumer start is pending leaves the
> delegated consumer running on a non-leader node
> ----------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-24583
> URL: https://issues.apache.org/jira/browse/CAMEL-24583
> Project: Camel
> Issue Type: Bug
> Components: camel-master
> Affects Versions: 4.18.3, 4.20.0, 4.21.0, 4.22.0
> Reporter: Henrik
> Priority: Major
>
> {{MasterConsumer}} starts the delegated consumer asynchronously, from a
> {{BackgroundTask}} scheduled one second after the leadership-taken event. A
> leadership-lost event that arrives inside that window is discarded, and the
> task then starts the consumer on a node that is no longer the leader. Nothing
> stops it afterwards, because no further leadership event is coming: the node
> is already a follower.
> The result is two active consumers on an endpoint that {{master:}} exists to
> keep single-consumer.
> Line numbers below are 4.22.0.
> # {{onLeadershipTaken()}} (line 154) holds {{lock}} only long enough to
> schedule the task (line 166). {{delegatedConsumer}} is still null when it
> returns.
> # The scheduled body's only guard is {{if (!isRunAllowed()) return false;}}
> (line 167). {{isRunAllowed()}} is the consumer's own {{BaseService}} status,
> that is, whether the route is running. It says nothing about leadership.
> # {{LeadershipListener.leadershipChanged()}} (line 235) dispatches to
> {{onLeadershipLost()}} only {{else if (delegatedConsumer != null)}} (line
> 245). While the start is pending that field is null, so the lost event is
> dropped and nothing cancels the pending task.
> # One second later (line 147,
> {{{}withInitialDelay(Duration.ofSeconds(1)){}}}, not configurable) the task
> creates and starts the delegated consumer. The node is a follower.
> h2. Impact
> Observed in production on Camel 4.21.0 with {{KubernetesClusterService}} and
> a google-pubsub delegate. A three second leadership flap produced repeated
> taken/lost cycles on the same endpoint, at sub-second intervals well inside
> the one second start delay:
>
>
> {code}
> 16:01:27.567 Leadership lost. Consumer stopped:
> google-pubsub://<project>:<Queue>
>
> 16:01:28.671 Leadership taken. Attempt #1 to start consumer:
> google-pubsub://<project>:<Queue>
>
> 16:01:29.932 Stopping subscribers for <project>/<Queue>
>
>
> {code}
>
>
> That is the precondition for the race: a lost event arriving while a start is
> pending is dropped, because the listener guards it on delegatedConsumer !=
> null.
> h2. Reproducer
> Fake the cluster view so leadership can be flapped on demand, then assert
> that the delegated consumer is never started:
> {code:java}
> // view.getLocalMember().isLeader() returns leader.get()
> // masterEndpoint.getEndpoint() returns a mock Endpoint whose
> createConsumer() is observable
> consumer = new MasterConsumer(masterEndpoint, processor, clusterService);
> consumer.start();
> leader.set(true);
> leadershipListener.leadershipChanged(view, localMember); // taken
> leader.set(false);
> leadershipListener.leadershipChanged(view, localMember); // lost, start
> still pending
> Thread.sleep(3000); // outlast the 1s
> initial delay
> verify(delegatedEndpoint, never()).createConsumer(any()); // fails: the
> consumer was started
> {code}
> h2. Two related observations in the same class
> * {{BackgroundTask.schedule()}} returns the {{scheduleWithFixedDelay}}
> future, but {{MasterConsumer }}neither retains nor cancels it. The task only
> short-circuits through its own latch ({{{}BackgroundTask.runTaskWrapper{}}},
> {{{}if (latch.getCount() == 0) return;{}}}), so every leadership change
> leaves a repeating no-op task on the "Leadership" pool for the life of the
> route.
> * If {{ServiceHelper.startService()}} throws, {{delegatedConsumer}} has
> already been assigned and is left non-null but unstarted. Once the retry
> budget is exhausted, {{{}onLeadershipTaken(){}}}'s {{if (delegatedConsumer !=
> null) return;}} early-return means a later leadership-taken event schedules
> nothing, and the node stays nominally leader while consuming nothing until it
> restarts.
> h2. Proposed fix
> * Track the leadership taken/lost state, guarded by the same lock the start
> and the stop take, and re-check it inside the scheduled task before creating
> the delegated consumer.
> * Read the leadership inside that lock in the listener, and dispatch the
> lost event unconditionally, so a lost event during a pending start cancels it
> instead of being dropped.
> * Retain the {{Future}} from {{BackgroundTask.schedule()}} and cancel it on
> leadership lost, on a successful start, and on stop.
> * Publish {{delegatedConsumer}} only after it has started, so a failed
> attempt cannot make every later leadership term a no-op.
> A patch with tests is ready; the {{BackgroundTask}} half (a scheduled task
> should unschedule itself once it has completed or run out of budget) is
> proposed as a separate camel-support issue.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
