From 93d9e1eb4e3224cb5130f6b309a83924d5687386 Mon Sep 17 00:00:00 2001 From: Prashant Bhardwaj Date: Fri, 10 Jul 2026 01:51:11 +0530 Subject: [PATCH] [FLINK-40114][runtime] Don't erase leader information while a confirmation is pending Generated-by: Claude Code (Claude Opus 4.8) --- .../DefaultLeaderElectionService.java | 9 +- .../DefaultLeaderElectionServiceTest.java | 140 +++++++++++++++++- 2 files changed, 146 insertions(+), 3 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java index ebd1d70dc95f6..08bba01e1cb8e 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java @@ -517,10 +517,17 @@ private void notifyLeaderInformationChangeInternal( } if (confirmedLeaderInformation.isEmpty()) { + // No confirmed leader information for this component yet (e.g. leadership was + // (re-)acquired but the asynchronous confirmation hasn't completed). Leave the external + // storage untouched instead of deleting it: a pending confirmation will overwrite any + // stale value, whereas deleting here would drop the leader information permanently if + // the confirmation never completes. LOG.trace( - "Leader information changed while there's no confirmation available by the contender for component '{}', yet. Changed leader information {} will be reset.", + "Leader information for component '{}' changed while no confirmation is available yet (issued leader session ID: {}). Leaving the external storage untouched. Changed leader information: {}.", componentId, + issuedLeaderSessionID, LeaderElectionUtils.convertToString(externallyChangedLeaderInformation)); + return; } else if (externallyChangedLeaderInformation.isEmpty()) { LOG.debug( "Re-writing leader information ({}) for component '{}' to overwrite the empty leader information in the external storage.", diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java index 2b0719489924e..3292ed320f7e3 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java @@ -1105,6 +1105,138 @@ void testOldConfirmationWhileHavingLeadershipLost() throws Exception { }; } + /** + * A leader-information change event (e.g. a Kubernetes ConfigMap watch event) that arrives + * while a granted session's confirmation is still in flight must not erase the component's + * leader information from the external storage. {@code confirmLeadershipAsync} runs + * asynchronously on the {@code leadershipOperationExecutor} while {@code + * onLeaderInformationChange} is handled synchronously, so the change handler can observe an + * empty {@code confirmedLeaderInformation} and reset a stale external entry to empty before the + * confirmation has had a chance to run. + */ + @Test + void testLeaderInformationChangeDoesNotEraseComponentWithInflightConfirmation() + throws Exception { + final AtomicReference storedLeaderInformation = + new AtomicReference<>(); + new Context(storedLeaderInformation) { + { + runTestWithManuallyTriggeredEvents( + executor -> { + final String componentId = contenderContext0.componentId; + final UUID staleSessionId = UUID.randomUUID(); + final UUID currentSessionId = UUID.randomUUID(); + + // Acquire leadership for the current session but keep the confirmation + // in flight: the grant runnable runs, the auto-queued confirm does not. + grantLeadership(currentSessionId); + executor.trigger(); + + // The external storage still holds a stale entry from a previous + // session that wasn't cleaned up (e.g. during an API-server outage). + final LeaderInformation staleLeaderInformation = + LeaderInformation.known(staleSessionId, "stale-address"); + storedLeaderInformation.set( + LeaderInformationRegister.of( + componentId, staleLeaderInformation)); + + // A watch event delivers the stale entry while the confirmation for the + // current session is still queued on the leadershipOperationExecutor. + leaderElectionService.onLeaderInformationChange( + LeaderInformationRegister.of( + componentId, staleLeaderInformation)); + + assertThat( + storedLeaderInformation + .get() + .hasLeaderInformation(componentId)) + .as( + "The leader information for a component we lead must not be " + + "erased from the external storage while its confirmation " + + "is still in flight.") + .isTrue(); + + // Once the pending confirmation runs, the external storage converges to + // the current session's leader information. + executor.triggerAll(); + assertThat(storedLeaderInformation.get().forComponentId(componentId)) + .hasValue( + LeaderInformation.known( + currentSessionId, contenderContext0.address)); + }); + } + }; + } + + /** + * After leadership is lost and re-acquired with a new session, the previous session's + * still-present external entry must not be erased while the new session's confirmation is in + * flight. A component confirms one session, gets revoked (which clears the local confirmation + * but leaves the external entry in place), and is re-granted a new session whose confirmation + * is still pending when a change event replays the stale entry. + */ + @Test + void testLeaderInformationChangeDoesNotEraseComponentAfterReacquisition() throws Exception { + final AtomicReference storedLeaderInformation = + new AtomicReference<>(); + new Context(storedLeaderInformation) { + { + runTestWithManuallyTriggeredEvents( + executor -> { + final String componentId = contenderContext0.componentId; + final UUID firstSessionId = UUID.randomUUID(); + final UUID secondSessionId = UUID.randomUUID(); + + // First reign: grant and let the confirmation land so the external + // storage holds the first session's leader information. + grantLeadership(firstSessionId); + executor.triggerAll(); + assertThat( + storedLeaderInformation + .get() + .hasLeaderInformation(componentId)) + .as("precondition: the first session's entry was written") + .isTrue(); + + // Leadership is lost: the local confirmation is cleared but the + // external + // storage keeps the stale entry (not cleaned up during the outage). + revokeLeadership(); + executor.triggerAll(); + + // Leadership is re-acquired with a new session; the confirmation for + // the + // new session is queued but kept in flight. + grantLeadership(secondSessionId); + executor.trigger(); + + // A watch event replays the stale (first-session) entry while the new + // session's confirmation is still pending. + leaderElectionService.onLeaderInformationChange( + storedLeaderInformation.get()); + + assertThat( + storedLeaderInformation + .get() + .hasLeaderInformation(componentId)) + .as( + "The leader information must not be erased after " + + "re-acquisition while the new session's confirmation " + + "is still in flight.") + .isTrue(); + + // Once the pending confirmation runs, the external storage converges to + // the new session's leader information. + executor.triggerAll(); + assertThat(storedLeaderInformation.get().forComponentId(componentId)) + .hasValue( + LeaderInformation.known( + secondSessionId, contenderContext0.address)); + }); + } + }; + } + @Test void testErrorForwarding() throws Exception { new Context() { @@ -1201,9 +1333,13 @@ private void testLeaderEventDoesNotBlockLeaderInformationChangeEventHandling( contenderContext0.componentId, storedLeaderInformation.get()); - assertThat(storedLeaderInformation.get().hasNoLeaderInformation()) + assertThat( + storedLeaderInformation + .get() + .hasLeaderInformation( + contenderContext0.componentId)) .as( - "The blocked leadership grant event shouldn't have blocked the processing of the LeaderInformation change event.") + "A queued leadership grant event shouldn't block the LeaderInformation change event; while the grant is pending (no confirmation available yet) the change handler must not erase the still-unconfirmed leader information.") .isTrue(); }); }