Henrik created CAMEL-24583:
------------------------------
Summary: 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.22.0, 4.21.0, 4.20.0, 4.18.3
Reporter: Henrik
{{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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
