[
https://issues.apache.org/jira/browse/CAMEL-24583?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Henrik updated CAMEL-24583:
---------------------------
Patch Info: Patch Available
> 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
> 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)
