[
https://issues.apache.org/jira/browse/FLINK-40114?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Prashant Bhardwaj updated FLINK-40114:
--------------------------------------
Description:
h3. Summary
Since FLINK-36451 made leadership confirmation asynchronous, a component's
leader information can be *permanently deleted* from the HA backend (e.g. the
Kubernetes leader ConfigMap) after a leadership loss and re-acquisition.
h3. Root cause
Confirmation runs asynchronously ({{{}confirmLeadershipAsync{}}} on the
single-threaded {{{}leadershipOperationExecutor{}}}), while the
leader-information-change handler ({{{}onLeaderInformationChange{}}} →
{{{}notifyLeaderInformationChangeInternal{}}}) still runs synchronously on the
driver's watch thread. After a loss + re-acquisition, a component's local
confirmation is cleared but its stale entry remains in the HA store. If a
change event is handled before the re-granted session's confirmation lands, the
handler takes the [empty-confirmation reset
branch|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java#L519]
and publishes empty, which on Kubernetes [removes the
key|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriver.java#L161].
Because the self-correcting path runs only once a confirmation exists, if that
confirmation never lands — due to a transient getConfigMap() failure or session
churn — the deletion is permanent. Triggered by any API-server disruption
(outage, rolling upgrade, or intermittent 5xx/429 under load).
h3. Impact
The victim is whichever component's latest-session confirmation fails to land:
* {{restserver}} erased → REST endpoint cannot be resolved (operator/clients
cannot reach the cluster).
* {{{}dispatcher{}}}/{{{}jobmanager{}}} erased → no job submission, no
TaskManager registration.
The JobManager keeps running and believes it is a healthy leader, so it is
silent and does not self-recover.
h3. Reproduction
Reproducible deterministically as a unit test by driving
{{DefaultLeaderElectionService}} with a manually-triggered
{{leadershipOperationExecutor}} so the confirmation can be held mid-flight: a
component wins leadership and confirms a session, loses leadership (clearing
the local confirmation but leaving the stale entry in the store), then wins
again; while the new session's confirmation is still pending, a
leader-information-change event carrying the stale entry drops the component's
key.
was:
h3. Summary
Since FLINK-36451 made leadership confirmation asynchronous, a race between the
asynchronous confirmation
([{{DefaultLeaderElectionService#confirmLeadershipAsync}}|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java#L316],
executed on the single-threaded {{{}leadershipOperationExecutor{}}}) and the
_synchronous_ leader-information-change handler
([{{onLeaderInformationChange}}|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java#L620],
executed directly on the driver's watch thread) can cause a component's leader
information to be *permanently deleted* from the HA backend (e.g. the
Kubernetes leader ConfigMap) after a leadership loss and re-acquisition.
h3. Root cause
{{confirmLeadershipAsync}} sets {{confirmedLeaderInformation}} and publishes
the leader entry asynchronously, whereas {{onLeaderInformationChange}} →
[{{notifyLeaderInformationChangeInternal}}|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java#L502]
runs synchronously under {{lock}} on the watch thread. When a component is
(re-)granted leadership but its confirmation has not yet populated
{{{}confirmedLeaderInformation{}}}, and the external store still holds a stale
entry from a previous session, the change handler takes the
[{{confirmedLeaderInformation.isEmpty()}}
branch|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java#L519]
and [publishes
_empty_|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java#L538]
for that component. On Kubernetes this maps to
[{{data.remove(key)}}|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriver.java#L161]
in
[{{KubernetesLeaderElectionDriver}}|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriver.java#L126],
deleting the leader key.
Because the self-correcting ("Correcting"/"Re-writing") paths only run when
{{confirmedLeaderInformation}} is non-empty, if that component's confirmation
never lands the deletion is {*}permanent{*}. A confirmation fails to land when:
* a transient failure is thrown from
[{{hasLeadershipInternal}}|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java#L380]
→ {{hasLeadership()}} → {{getConfigMap()}} — the returned future completes
exceptionally and is neither retried nor surfaced; or
* session churn occurs — the confirmation targets a session that is no longer
the issued one and no-ops.
There are two symmetric windows in which the erasure can occur while we hold
the lock:
* {_}confirm-pending{_}: {{issuedLeaderSessionID != null}} but the
confirmation runnable is still queued or has already failed.
* {_}grant-pending{_}: the lock is already held (so {{hasLeadership()}} is
true) but
[{{onGrantLeadership}}|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java#L598]
has not been processed yet, so {{{}issuedLeaderSessionID == null{}}}. The
lock-acquire annotation write itself triggers an
[{{onModified}}|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriver.java#L216]
watch event carrying the stale entries.
h3. Trigger
It is triggered by any API-server disruption that both (a) causes a lease-renew
miss (leadership loss + re-acquisition) and (b) makes a component's
confirmation fail to land. Examples:
* a full API-server outage;
* a control-plane rolling upgrade;
* *intermittent 5xx / 429 throttling under load* — a single badly-timed
{{{}5xx{}}}/{{{}429{}}} on the confirmation's {{getConfigMap()}} GET is enough
to make it throw (and be swallowed), and repeated renew failures/latency can
cause the loss + re-acquire.
The stale entry is essentially always present in a loss + re-acquisition:
revocation clears the local confirmation but leaves the external entry in place
(and a fresh failover inherits the previous leader's entry). What actually
varies is (a) whether a confirmation fails to land and (b) whether a watch
event fires during the unconfirmed window. During a sustained disruption, both
are near-certain and hit many JobManagers at once. Under merely intermittent
errors, the two must line up in one brief window, so it is lower-probability
per re-acquisition but non-zero and routine at fleet scale. The single-threaded
{{leadershipOperationExecutor}} is an amplifier: slow (even non-failing)
{{getConfigMap()}} calls stall confirmations and widen the window.
h3. Impact
The victim is whichever component's latest-session confirmation fails to land.
Consequences by component:
* {{restserver}} erased → REST endpoint cannot be resolved (operator/clients
cannot reach the cluster).
* {{{}dispatcher{}}}/{{{}jobmanager{}}} erased → no job submission, no
TaskManager registration.
The JobManager keeps running and believes it is a healthy leader, so this is
silent and doe
h3. Reproduction
The race can be reproduced deterministically as a unit test by driving
{{DefaultLeaderElectionService}} with a manually-triggered
{{{}leadershipOperationExecutor{}}}, so the async confirmation can be held
mid-flight. The scenario mirrors an outage recovery: a component wins
leadership and confirms a session, clears the local confirmation but leaves the
now-stale entry in the HA store), then wins again. While the new session's
confirmation is still pending, a leader-information-change event arrives
carrying the stale entry, and the component's key is dropped from the store.
The added tests in {{DefaultLeaderElectionServiceTest}} cover this, along with
the session-churn and grant-pent the entry both survives the race and recovers
once the pending confirmation completes.
> Leader information permanently erased when a change event races a pending
> confirmation
> --------------------------------------------------------------------------------------
>
> Key: FLINK-40114
> URL: https://issues.apache.org/jira/browse/FLINK-40114
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Coordination
> Affects Versions: 2.4.0
> Reporter: Prashant Bhardwaj
> Priority: Major
>
> h3. Summary
> Since FLINK-36451 made leadership confirmation asynchronous, a component's
> leader information can be *permanently deleted* from the HA backend (e.g. the
> Kubernetes leader ConfigMap) after a leadership loss and re-acquisition.
> h3. Root cause
> Confirmation runs asynchronously ({{{}confirmLeadershipAsync{}}} on the
> single-threaded {{{}leadershipOperationExecutor{}}}), while the
> leader-information-change handler ({{{}onLeaderInformationChange{}}} →
> {{{}notifyLeaderInformationChangeInternal{}}}) still runs synchronously on
> the driver's watch thread. After a loss + re-acquisition, a component's local
> confirmation is cleared but its stale entry remains in the HA store. If a
> change event is handled before the re-granted session's confirmation lands,
> the handler takes the [empty-confirmation reset
> branch|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java#L519]
> and publishes empty, which on Kubernetes [removes the
> key|https://github.com/apache/flink/blob/f9d81b2e11e5cff7469b523cc73ce34b1d28e5ab/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriver.java#L161].
> Because the self-correcting path runs only once a confirmation exists, if
> that confirmation never lands — due to a transient getConfigMap() failure or
> session churn — the deletion is permanent. Triggered by any API-server
> disruption (outage, rolling upgrade, or intermittent 5xx/429 under load).
> h3. Impact
> The victim is whichever component's latest-session confirmation fails to land:
> * {{restserver}} erased → REST endpoint cannot be resolved (operator/clients
> cannot reach the cluster).
> * {{{}dispatcher{}}}/{{{}jobmanager{}}} erased → no job submission, no
> TaskManager registration.
> The JobManager keeps running and believes it is a healthy leader, so it is
> silent and does not self-recover.
> h3. Reproduction
> Reproducible deterministically as a unit test by driving
> {{DefaultLeaderElectionService}} with a manually-triggered
> {{leadershipOperationExecutor}} so the confirmation can be held mid-flight: a
> component wins leadership and confirms a session, loses leadership (clearing
> the local confirmation but leaving the stale entry in the store), then wins
> again; while the new session's confirmation is still pending, a
> leader-information-change event carrying the stale entry drops the
> component's key.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
