You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2023/07/07 07:54:06 UTC
[flink] 01/02: [FLINK-31814][runtime] Enables Precondition that checks that the issuedLeaderSessionID is set on revoke processing
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c5fc72361f198692421dfb50e00e251b84e8d4cc
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Wed Jun 28 13:56:29 2023 +0200
[FLINK-31814][runtime] Enables Precondition that checks that the issuedLeaderSessionID is set on revoke processing
This Precondition can be enabled because the processing of revoke calls is not happening anymore during close. The event processing is protected by the running flag within DefaultLeaderElectionService which is set to false as part of the close mechanism.
Signed-off-by: Matthias Pohl <ma...@aiven.io>
---
.../DefaultLeaderElectionService.java | 6 +-
.../DefaultLeaderElectionServiceTest.java | 68 +++++++++++++---------
2 files changed, 45 insertions(+), 29 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 9e67c597518..eb3d3788075 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
@@ -424,9 +424,9 @@ public class DefaultLeaderElectionService extends AbstractLeaderElectionService
@GuardedBy("lock")
private void onRevokeLeadershipInternal() {
- // TODO: FLINK-31814 covers adding this Precondition
- // Preconditions.checkState(issuedLeaderSessionID != null,"The leadership should have
- // been revoked while having the leadership acquired.");
+ Preconditions.checkState(
+ issuedLeaderSessionID != null,
+ "The leadership should have been revoked while having the leadership acquired.");
if (!leaderContenderRegistry.isEmpty()) {
leaderContenderRegistry.forEach(this::notifyLeaderContenderOfLeadershipLoss);
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 8fb91264e8b..15aaffe99fe 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
@@ -36,6 +36,7 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.Function;
@@ -317,39 +318,54 @@ class DefaultLeaderElectionServiceTest {
};
}
- /**
- * Test to cover the issue described in FLINK-31814. This test could be removed after
- * FLINK-31814 is resolved.
- */
@Test
- void testOnRevokeCallWhileClosingService() throws Exception {
- final AtomicBoolean leadershipGranted = new AtomicBoolean();
- final TestingLeaderElectionDriver.Builder driverBuilder =
- TestingLeaderElectionDriver.newBuilder(leadershipGranted);
+ void testDelayedRevokeCallAfterContenderBeingDeregisteredAgain() throws Exception {
+ new Context() {
+ {
+ runTestWithManuallyTriggeredEvents(
+ executorService -> {
+ final UUID expectedSessionID = UUID.randomUUID();
+ grantLeadership(expectedSessionID);
+ executorService.trigger();
- final TestingLeaderElectionDriver.Factory driverFactory =
- new TestingLeaderElectionDriver.Factory(driverBuilder);
- try (final DefaultLeaderElectionService testInstance =
- new DefaultLeaderElectionService(
- driverFactory, fatalErrorHandlerExtension.getTestingFatalErrorHandler())) {
- driverBuilder.setCloseConsumer(lock -> testInstance.onRevokeLeadership());
- testInstance.startLeaderElectionBackend();
+ final LeaderElection leaderElection =
+ leaderElectionService.createLeaderElection("contender_id");
- leadershipGranted.set(true);
- testInstance.onGrantLeadership(UUID.randomUUID());
+ final AtomicInteger revokeCallCount = new AtomicInteger();
+ final LeaderContender contender =
+ TestingGenericLeaderContender.newBuilder()
+ .setRevokeLeadershipRunnable(
+ revokeCallCount::incrementAndGet)
+ .build();
+ leaderElection.startLeaderElection(contender);
+ executorService.trigger();
- final LeaderElection leaderElection =
- testInstance.createLeaderElection(createRandomContenderID());
- final TestingContender contender =
- new TestingContender("unused-address", leaderElection);
- contender.startLeaderElection();
+ assertThat(revokeCallCount)
+ .as("No revocation should have been triggered, yet.")
+ .hasValue(0);
- contender.waitForLeader();
+ revokeLeadership();
- leaderElection.close();
+ assertThat(revokeCallCount)
+ .as("A revocation was triggered but not processed, yet.")
+ .hasValue(0);
- contender.throwErrorIfPresent();
- }
+ leaderElection.close();
+
+ assertThat(revokeCallCount)
+ .as(
+ "A revocation should have been triggered and immediately processed through the close call.")
+ .hasValue(1);
+
+ executorService.triggerAll();
+
+ assertThat(revokeCallCount)
+ .as(
+ "The leadership revocation event that was triggered by the HA backend shouldn't have been forwarded to the contender, anymore.")
+ .hasValue(1);
+ });
+ }
+ };
}
@Test