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/09 16:11:00 UTC
[flink] 01/03: [FLINK-31837] Makes the driver instantiation being performed lazily as soon as the first contender is registered
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 d2e6b94b374a40551a5cb1a81af2683f4a9ce86d
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Fri Jun 30 15:14:58 2023 +0200
[FLINK-31837] Makes the driver instantiation being performed lazily as soon as the first contender is registered
- the driver's lifecycle is tied to the leaderContenderRegistry: it's only instantiated if there are contenders registered
- the service's lifecycle takes care of the internally used leadershipOperationExecutor
---
.../highavailability/AbstractHaServices.java | 50 +--
.../highavailability/HighAvailabilityServices.java | 8 +-
.../leaderelection/DefaultLeaderElection.java | 2 +-
.../DefaultLeaderElectionService.java | 124 +++++---
.../JobMasterServiceLeadershipRunnerTest.java | 1 -
.../DefaultLeaderElectionServiceTest.java | 351 +++++++++++++--------
.../runtime/leaderelection/LeaderElectionTest.java | 1 -
.../TestingLeaderElectionDriver.java | 4 +
...KeeperLeaderElectionConnectionHandlingTest.java | 1 -
.../ZooKeeperLeaderElectionTest.java | 18 +-
10 files changed, 323 insertions(+), 237 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java
index d57e3d94dc4..bcca7f3e866 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java
@@ -33,8 +33,6 @@ import org.apache.flink.util.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.concurrent.GuardedBy;
-
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
@@ -70,12 +68,7 @@ public abstract class AbstractHaServices implements HighAvailabilityServices {
private final JobResultStore jobResultStore;
- private final LeaderElectionDriverFactory driverFactory;
-
- private final Object lock = new Object();
-
- @GuardedBy("lock")
- private DefaultLeaderElectionService leaderElectionService;
+ private final DefaultLeaderElectionService leaderElectionService;
protected AbstractHaServices(
Configuration config,
@@ -89,7 +82,7 @@ public abstract class AbstractHaServices implements HighAvailabilityServices {
this.blobStoreService = checkNotNull(blobStoreService);
this.jobResultStore = checkNotNull(jobResultStore);
- this.driverFactory = driverFactory;
+ this.leaderElectionService = new DefaultLeaderElectionService(driverFactory);
}
@Override
@@ -119,23 +112,23 @@ public abstract class AbstractHaServices implements HighAvailabilityServices {
}
@Override
- public LeaderElection getResourceManagerLeaderElection() throws Exception {
- return createLeaderElection(getLeaderPathForResourceManager());
+ public LeaderElection getResourceManagerLeaderElection() {
+ return leaderElectionService.createLeaderElection(getLeaderPathForResourceManager());
}
@Override
- public LeaderElection getDispatcherLeaderElection() throws Exception {
- return createLeaderElection(getLeaderPathForDispatcher());
+ public LeaderElection getDispatcherLeaderElection() {
+ return leaderElectionService.createLeaderElection(getLeaderPathForDispatcher());
}
@Override
- public LeaderElection getJobManagerLeaderElection(JobID jobID) throws Exception {
- return createLeaderElection(getLeaderPathForJobManager(jobID));
+ public LeaderElection getJobManagerLeaderElection(JobID jobID) {
+ return leaderElectionService.createLeaderElection(getLeaderPathForJobManager(jobID));
}
@Override
- public LeaderElection getClusterRestEndpointLeaderElection() throws Exception {
- return createLeaderElection(getLeaderPathForRestServer());
+ public LeaderElection getClusterRestEndpointLeaderElection() {
+ return leaderElectionService.createLeaderElection(getLeaderPathForRestServer());
}
@Override
@@ -169,10 +162,8 @@ public abstract class AbstractHaServices implements HighAvailabilityServices {
}
try {
- synchronized (lock) {
- if (leaderElectionService != null) {
- leaderElectionService.close();
- }
+ if (leaderElectionService != null) {
+ leaderElectionService.close();
}
} catch (Throwable t) {
exception = ExceptionUtils.firstOrSuppressed(t, exception);
@@ -206,10 +197,8 @@ public abstract class AbstractHaServices implements HighAvailabilityServices {
}
try {
- synchronized (lock) {
- if (leaderElectionService != null) {
- leaderElectionService.close();
- }
+ if (leaderElectionService != null) {
+ leaderElectionService.close();
}
} catch (Throwable t) {
exception = ExceptionUtils.firstOrSuppressed(t, exception);
@@ -257,17 +246,6 @@ public abstract class AbstractHaServices implements HighAvailabilityServices {
executor);
}
- private LeaderElection createLeaderElection(String leaderName) throws Exception {
- synchronized (lock) {
- if (leaderElectionService == null) {
- leaderElectionService = new DefaultLeaderElectionService(driverFactory);
- leaderElectionService.startLeaderElectionBackend();
- }
-
- return leaderElectionService.createLeaderElection(leaderName);
- }
- }
-
/**
* Create leader retrieval service with specified leaderName.
*
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
index 5b1b91afea3..97ef9a4b1fe 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
@@ -121,13 +121,13 @@ public interface HighAvailabilityServices
}
/** Gets the {@link LeaderElection} for the cluster's resource manager. */
- LeaderElection getResourceManagerLeaderElection() throws Exception;
+ LeaderElection getResourceManagerLeaderElection();
/** Gets the {@link LeaderElection} for the cluster's dispatcher. */
- LeaderElection getDispatcherLeaderElection() throws Exception;
+ LeaderElection getDispatcherLeaderElection();
/** Gets the {@link LeaderElection} for the job with the given {@link JobID}. */
- LeaderElection getJobManagerLeaderElection(JobID jobID) throws Exception;
+ LeaderElection getJobManagerLeaderElection(JobID jobID);
/**
* Gets the {@link LeaderElection} for the cluster's rest endpoint.
@@ -176,7 +176,7 @@ public interface HighAvailabilityServices
BlobStore createBlobStore() throws IOException;
/** Gets the {@link LeaderElection} for the cluster's rest endpoint. */
- default LeaderElection getClusterRestEndpointLeaderElection() throws Exception {
+ default LeaderElection getClusterRestEndpointLeaderElection() {
// for backwards compatibility we delegate to getWebMonitorLeaderElectionService
// all implementations of this interface should override
// getClusterRestEndpointLeaderElectionService, though
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElection.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElection.java
index a0867403f01..3a694c5ec2d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElection.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElection.java
@@ -74,7 +74,7 @@ class DefaultLeaderElection implements LeaderElection {
* Removes the {@code LeaderContender} from the {@code ParentService} that is associated
* with the {@code contenderID}.
*/
- abstract void remove(String contenderID);
+ abstract void remove(String contenderID) throws Exception;
/**
* Confirms the leadership with the {@code leaderSessionID} and {@code leaderAddress} for
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 e3492e81fd4..83b0504c234 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
@@ -90,17 +90,13 @@ public class DefaultLeaderElectionService extends DefaultLeaderElection.ParentSe
private boolean running;
/**
- * {@code leaderElectionDriver} being {@code null} indicates that the connection to the
- * LeaderElection backend isn't established, yet. See {@link #startLeaderElectionBackend()} and
- * {@link #close()} for lifecycle management. The lifecycle of the driver should have been
- * established before registering a {@link LeaderContender} and stopped after the contender has
- * been removed.
- *
- * <p>{@code @Nullable} isn't used here to avoid having multiple warnings spread over this class
- * in a supporting IDE.
- *
- * <p>The driver is guarded by this instance's {@link #running} state.
+ * The driver's lifecycle is bound to the {@link #leaderContenderRegistry}: {@code
+ * leaderElectionDriver} is {@code null} if no contender is registered: A new driver is created
+ * as soon as the first contender is added to the empty {@code leaderContenderRegistry}. Only
+ * then, a connection to the {@code DefaultLeaderElectionService} backend is established. The
+ * service resets and closes the driver with the removal of the last contender.
*/
+ @GuardedBy("lock")
private LeaderElectionDriver leaderElectionDriver;
/**
@@ -151,12 +147,15 @@ public class DefaultLeaderElectionService extends DefaultLeaderElection.ParentSe
this.leadershipOperationExecutor = Preconditions.checkNotNull(leadershipOperationExecutor);
- this.running = false;
+ this.running = true;
}
@Override
public LeaderElection createLeaderElection(String contenderID) {
synchronized (lock) {
+ Preconditions.checkState(
+ !leadershipOperationExecutor.isShutdown(),
+ "The service was already closed and cannot be reused.");
Preconditions.checkState(
!leaderContenderRegistry.containsKey(contenderID),
"There is no contender already registered under the passed contender ID '%s'.",
@@ -165,28 +164,20 @@ public class DefaultLeaderElectionService extends DefaultLeaderElection.ParentSe
}
}
- /**
- * Starts the leader election process. This method has to be called before registering a {@link
- * LeaderContender}. This method could be moved into the {@code DefaultLeaderElectionService}'s
- * constructor with FLINK-31837.
- */
- public void startLeaderElectionBackend() throws Exception {
- synchronized (lock) {
- Preconditions.checkState(
- leaderContenderRegistry.isEmpty(),
- "No LeaderContender should have been registered, yet.");
- Preconditions.checkState(
- leaderElectionDriver == null,
- "This DefaultLeaderElectionService cannot be reused. Calling startLeaderElectionBackend can only be called once to establish the connection to the HA backend.");
-
- running = true;
+ @GuardedBy("lock")
+ private void createLeaderElectionDriver() throws Exception {
+ Preconditions.checkState(
+ leaderContenderRegistry.isEmpty(),
+ "No LeaderContender should have been registered, yet.");
+ Preconditions.checkState(
+ leaderElectionDriver == null,
+ "This DefaultLeaderElectionService cannot be reused. Calling startLeaderElectionBackend can only be called once to establish the connection to the HA backend.");
- leaderElectionDriver = leaderElectionDriverFactory.create(this);
+ leaderElectionDriver = leaderElectionDriverFactory.create(this);
- LOG.info(
- "A connection to the HA backend was established through LeaderElectionDriver {}.",
- leaderElectionDriver);
- }
+ LOG.info(
+ "A connection to the HA backend was established through LeaderElectionDriver {}.",
+ leaderElectionDriver);
}
@Override
@@ -199,6 +190,10 @@ public class DefaultLeaderElectionService extends DefaultLeaderElection.ParentSe
running,
"The DefaultLeaderElectionService should have established a connection to the backend before it's started.");
+ if (leaderElectionDriver == null) {
+ createLeaderElectionDriver();
+ }
+
Preconditions.checkState(
leaderContenderRegistry.put(contenderID, contender) == null,
"There is no contender already registered under the passed contender ID '%s'.",
@@ -221,7 +216,8 @@ public class DefaultLeaderElectionService extends DefaultLeaderElection.ParentSe
}
@Override
- protected final void remove(String contenderID) {
+ protected final void remove(String contenderID) throws Exception {
+ AutoCloseable driverToClose = null;
synchronized (lock) {
if (!leaderContenderRegistry.containsKey(contenderID)) {
LOG.debug(
@@ -229,6 +225,9 @@ public class DefaultLeaderElectionService extends DefaultLeaderElection.ParentSe
contenderID);
return;
}
+ Preconditions.checkState(
+ leaderElectionDriver != null,
+ "The LeaderElectionDriver should be instantiated.");
LOG.info(
"Deregistering contender with ID '{}' from the DefaultLeaderElectionService.",
@@ -260,30 +259,54 @@ public class DefaultLeaderElectionService extends DefaultLeaderElection.ParentSe
"Contender registered under contenderID '{}' is deregistered while the service doesn't have the leadership acquired. No cleanup necessary.",
contenderID);
}
+
+ if (leaderContenderRegistry.isEmpty()) {
+ driverToClose = deregisterDriver();
+ }
+ }
+
+ if (driverToClose != null) {
+ driverToClose.close();
}
}
+ /**
+ * Returns the driver as an {@link AutoCloseable} for the sake of closing the driver outside of
+ * the lock.
+ */
+ @GuardedBy("lock")
+ private AutoCloseable deregisterDriver() {
+ Preconditions.checkState(
+ leaderContenderRegistry.isEmpty(),
+ "No contender should be registered when deregistering the driver.");
+ Preconditions.checkState(
+ leaderElectionDriver != null,
+ "There should be a driver instantiated that's ready to be closed.");
+
+ issuedLeaderSessionID = null;
+ final AutoCloseable driverToClose = leaderElectionDriver;
+ leaderElectionDriver = null;
+
+ return driverToClose;
+ }
+
@Override
public void close() throws Exception {
synchronized (lock) {
- Preconditions.checkState(
- leaderElectionDriver != null, "The HA backend wasn't initialized.");
-
Preconditions.checkState(
leaderContenderRegistry.isEmpty(),
"The DefaultLeaderElectionService should have been stopped before closing the instance.");
-
- issuedLeaderSessionID = null;
+ Preconditions.checkState(
+ leaderElectionDriver == null, "The driver should have been closed.");
if (running) {
running = false;
} else {
LOG.debug("The HA backend connection isn't established. No actions taken.");
+ return;
}
}
- leaderElectionDriver.close();
-
// interrupt any outstanding events
final List<Runnable> outstandingEventHandlingCalls =
leadershipOperationExecutor.shutdownNow();
@@ -308,6 +331,9 @@ public class DefaultLeaderElectionService extends DefaultLeaderElection.ParentSe
synchronized (lock) {
if (hasLeadership(contenderID, leaderSessionID)) {
+ Preconditions.checkState(
+ leaderElectionDriver != null,
+ "The leadership check should only return true if a driver is instantiated.");
Preconditions.checkState(
!confirmedLeaderInformation.hasLeaderInformation(contenderID),
"No confirmation should have happened, yet.");
@@ -468,6 +494,12 @@ public class DefaultLeaderElectionService extends DefaultLeaderElection.ParentSe
String contenderID,
LeaderInformation externallyChangedLeaderInformation,
LeaderInformation confirmedLeaderInformation) {
+ if (leaderElectionDriver == null) {
+ LOG.debug(
+ "The LeaderElectionDriver was disconnected. Any incoming events will be ignored.");
+ return;
+ }
+
if (confirmedLeaderInformation.equals(externallyChangedLeaderInformation)) {
LOG.trace(
"LeaderInformation change event received but changed LeaderInformation actually matches the locally confirmed one: {}",
@@ -505,15 +537,21 @@ public class DefaultLeaderElectionService extends DefaultLeaderElection.ParentSe
CompletableFuture.runAsync(
() -> {
synchronized (lock) {
- if (running) {
+ if (!running) {
LOG.debug(
- "Processing '{}' event.",
+ "Processing '{}' event omitted due to the service not being in running state, anymore.",
leaderElectionEventName);
- callback.run();
+ } else if (leaderElectionDriver == null) {
+ Preconditions.checkState(
+ leaderContenderRegistry.isEmpty(),
+ "All contenders should be deregistered when the driver is removed.");
+ LOG.debug(
+ "All contenders have been deregistered and the driver was shut down. Any incoming leadership event will be ignored.");
} else {
LOG.debug(
- "Processing '{}' event omitted due to the service not being in running state, anymore.",
+ "Processing '{}' event.",
leaderElectionEventName);
+ callback.run();
}
}
},
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java
index 3b2f8b6b6d0..3915bf03909 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java
@@ -692,7 +692,6 @@ class JobMasterServiceLeadershipRunnerTest {
// in connection with the DefaultLeaderElectionService generates the nested locking
final DefaultLeaderElectionService defaultLeaderElectionService =
new DefaultLeaderElectionService(driverFactory, fatalErrorHandler);
- defaultLeaderElectionService.startLeaderElectionBackend();
final TestingLeaderElectionDriver currentLeaderDriver =
driverFactory.assertAndGetOnlyCreatedDriver();
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 ab92ca08299..20575e6d86e 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
@@ -159,63 +159,155 @@ class DefaultLeaderElectionServiceTest {
final TestingLeaderElectionDriver.Factory driverFactory =
new TestingLeaderElectionDriver.Factory(driverBuilder);
- final DefaultLeaderElectionService testInstance =
+ try (final DefaultLeaderElectionService testInstance =
new DefaultLeaderElectionService(
- driverFactory, fatalErrorHandlerExtension.getTestingFatalErrorHandler());
- testInstance.startLeaderElectionBackend();
- final TestingLeaderElectionDriver driver = driverFactory.assertAndGetOnlyCreatedDriver();
+ driverFactory, fatalErrorHandlerExtension.getTestingFatalErrorHandler())) {
+
+ // creating the LeaderElection is necessary to instantiate the driver
+ final LeaderElection leaderElection = testInstance.createLeaderElection("contender-id");
+ leaderElection.startLeaderElection(TestingGenericLeaderContender.newBuilder().build());
+
+ final TestingLeaderElectionDriver driver =
+ driverFactory.assertAndGetOnlyCreatedDriver();
+
+ final Thread closeThread =
+ new Thread(
+ () -> {
+ try {
+ leaderElection.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ },
+ "CloseThread");
+
+ // triggers close that acquires the DefaultLeaderElectionService lock
+ closeThread.start();
+ closeReachedLatch.await();
+
+ final Thread grantThread =
+ new Thread(
+ () -> {
+ try {
+ // simulates the grant process being triggered from the HA
+ // backend's side where the same lock that is acquired during
+ // the driver's process is also acquired while handling a
+ // leadership event processing
+ driver.getLock().lock();
+ grantReachedLatch.trigger();
+ grantContinueLatch.awaitQuietly();
+
+ // grants leadership
+ leadershipGranted.set(true);
+ testInstance.onGrantLeadership(UUID.randomUUID());
+ } finally {
+ driver.getLock().unlock();
+ }
+ },
+ "GrantThread");
+
+ // triggers the service acquiring the leadership and, as a consequence, acquiring the
+ // driver's lock
+ grantThread.start();
+ grantReachedLatch.await();
+
+ // continue both processes which shouldn't result in a deadlock
+ grantContinueLatch.trigger();
+ closeContinueLatch.trigger();
+
+ closeThread.join();
+ grantThread.join();
+
+ FlinkAssertions.assertThatFuture(driverCloseTriggered).eventuallySucceeds();
+ }
+ }
- final Thread closeThread =
- new Thread(
- () -> {
- try {
- testInstance.close();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
+ @Test
+ void testLazyDriverInstantiation() throws Exception {
+ final AtomicBoolean driverCreated = new AtomicBoolean();
+ try (final DefaultLeaderElectionService testInstance =
+ new DefaultLeaderElectionService(
+ listener -> {
+ driverCreated.set(true);
+ return TestingLeaderElectionDriver.newNoOpBuilder().build(listener);
},
- "CloseThread");
+ fatalErrorHandlerExtension.getTestingFatalErrorHandler(),
+ Executors.newDirectExecutorService())) {
+ assertThat(driverCreated)
+ .as("The driver shouldn't have been created during service creation.")
+ .isFalse();
+
+ try (final LeaderElection leaderElection =
+ testInstance.createLeaderElection("contender-id")) {
+ assertThat(driverCreated)
+ .as(
+ "The driver shouldn't have been created during LeaderElection creation.")
+ .isFalse();
+
+ leaderElection.startLeaderElection(
+ TestingGenericLeaderContender.newBuilder().build());
+ assertThat(driverCreated)
+ .as(
+ "The driver should have been created when registering the contender in the LeaderElection.")
+ .isTrue();
+ }
+ }
+ }
- // triggers close that acquires the DefaultLeaderElectionService lock
- closeThread.start();
- closeReachedLatch.await();
+ @Test
+ void testReuseOfServiceIsRestricted() throws Exception {
+ final DefaultLeaderElectionService testInstance =
+ new DefaultLeaderElectionService(
+ new TestingLeaderElectionDriver.Factory(
+ TestingLeaderElectionDriver.newNoOpBuilder()));
- final Thread grantThread =
- new Thread(
- () -> {
- try {
- // simulates the grant process being triggered from the HA backend's
- // side where the same lock that is acquired during the driver's
- // process is also acquired while handling a leadership event
- // processing
- driver.getLock().lock();
- grantReachedLatch.trigger();
- grantContinueLatch.awaitQuietly();
-
- // grants leadership
- leadershipGranted.set(true);
- testInstance.onGrantLeadership(UUID.randomUUID());
- } finally {
- driver.getLock().unlock();
- }
- },
- "GrantThread");
+ // The driver hasn't started, yet, which prevents the service from going into running state.
+ // This results in the close method not having any effect.
+ testInstance.close();
- // triggers the service acquiring the leadership and, as a consequence, acquiring the
- // driver's lock
- grantThread.start();
- grantReachedLatch.await();
+ assertThatThrownBy(() -> testInstance.createLeaderElection("contender-id"))
+ .as(
+ "Registering a contender on a closed service should have resulted in an IllegalStateException.")
+ .isInstanceOf(IllegalStateException.class);
+ }
- // continue both processes which shouldn't result in a deadlock
- grantContinueLatch.trigger();
- closeContinueLatch.trigger();
+ /**
+ * The leadership can be granted as soon as the driver is instantiated. We need to make sure
+ * that the event is still handled properly while registering the contender.
+ */
+ @Test
+ void testMultipleDriverCreations() throws Exception {
+ final AtomicInteger closeCount = new AtomicInteger(0);
+ final TestingLeaderElectionDriver.Factory driverFactory =
+ new TestingLeaderElectionDriver.Factory(
+ TestingLeaderElectionDriver.newNoOpBuilder()
+ .setCloseConsumer(ignoredLock -> closeCount.incrementAndGet()));
- closeThread.join();
- grantThread.join();
+ try (final DefaultLeaderElectionService testInstance =
+ new DefaultLeaderElectionService(driverFactory)) {
+
+ final String contenderID = "contender_id";
+ final int numberOfStartCloseSessions = 2;
+ for (int i = 1; i <= numberOfStartCloseSessions; i++) {
+ assertThat(driverFactory.getCreatedDriverCount()).isEqualTo(i - 1);
+ assertThat(closeCount).hasValue(i - 1);
+
+ try (final LeaderElection leaderElection =
+ testInstance.createLeaderElection(contenderID)) {
+ leaderElection.startLeaderElection(
+ TestingGenericLeaderContender.newBuilder().build());
+ }
- FlinkAssertions.assertThatFuture(driverCloseTriggered).eventuallySucceeds();
+ assertThat(driverFactory.getCreatedDriverCount()).isEqualTo(i);
+ assertThat(closeCount).hasValue(i);
+ }
+ }
}
+ /**
+ * The leadership can be granted as soon as the driver is instantiated. We need to make sure
+ * that the event is still handled properly while registering the contender.
+ */
@Test
void testGrantCallWhileInstantiatingDriver() throws Exception {
final UUID expectedLeaderSessionID = UUID.randomUUID();
@@ -227,7 +319,6 @@ class DefaultLeaderElectionServiceTest {
},
fatalErrorHandlerExtension.getTestingFatalErrorHandler(),
Executors.newDirectExecutorService())) {
- testInstance.startLeaderElectionBackend();
final LeaderElection leaderElection =
testInstance.createLeaderElection(createRandomContenderID());
@@ -243,78 +334,89 @@ class DefaultLeaderElectionServiceTest {
@Test
void testDelayedGrantCallAfterContenderRegistration() throws Exception {
- new Context() {
- {
- runTestWithManuallyTriggeredEvents(
- executorService -> {
- // we need to close to deregister the contender that was already
- // registered to the service
- closeLeaderElectionInBothContexts();
-
- final UUID expectedSessionID = UUID.randomUUID();
- grantLeadership(expectedSessionID);
-
- applyToBothContenderContexts(
- ctx -> {
- try (LeaderElection anotherLeaderElection =
- leaderElectionService.createLeaderElection(
- ctx.contenderID)) {
- final TestingContender testingContender =
- new TestingContender(
- ctx.address, anotherLeaderElection);
- testingContender.startLeaderElection();
-
- assertThat(testingContender.getLeaderSessionID())
- .as(
- "Leadership grant was not forwarded to the contender, yet.")
- .isNull();
-
- executorService.trigger();
-
- assertThat(testingContender.getLeaderSessionID())
- .as(
- "Leadership grant is actually forwarded to the service.")
- .isEqualTo(expectedSessionID);
-
- testingContender.waitForLeader();
- }
- });
- });
+ final TestingLeaderElectionDriver.Factory driverFactory =
+ new TestingLeaderElectionDriver.Factory(
+ TestingLeaderElectionDriver.newNoOpBuilder());
+ final ManuallyTriggeredScheduledExecutorService leaderEventOperationExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+ try (final DefaultLeaderElectionService leaderElectionService =
+ new DefaultLeaderElectionService(
+ driverFactory,
+ fatalErrorHandlerExtension.getTestingFatalErrorHandler(),
+ leaderEventOperationExecutor)) {
+
+ final AtomicBoolean firstContenderReceivedGrant = new AtomicBoolean(false);
+ final LeaderContender firstContender =
+ TestingGenericLeaderContender.newBuilder()
+ .setGrantLeadershipConsumer(
+ ignoredSessionID -> firstContenderReceivedGrant.set(true))
+ .build();
+
+ final AtomicBoolean secondContenderReceivedGrant = new AtomicBoolean(false);
+ final LeaderContender secondContender =
+ TestingGenericLeaderContender.newBuilder()
+ .setGrantLeadershipConsumer(
+ ignoredSessionID -> secondContenderReceivedGrant.set(true))
+ .build();
+ try (final LeaderElection firstLeaderElection =
+ leaderElectionService.createLeaderElection("contender_id_0")) {
+ firstLeaderElection.startLeaderElection(firstContender);
+
+ assertThat(driverFactory.getCreatedDriverCount())
+ .as(
+ "A single driver should have been created when registering the contender.")
+ .isEqualTo(1);
+ leaderElectionService.onGrantLeadership(UUID.randomUUID());
+ assertThat(firstContenderReceivedGrant).isFalse();
+
+ try (final LeaderElection secondLeaderElection =
+ leaderElectionService.createLeaderElection("contender_id_1")) {
+ secondLeaderElection.startLeaderElection(secondContender);
+
+ assertThat(secondContenderReceivedGrant).isFalse();
+
+ leaderEventOperationExecutor.trigger();
+ assertThat(firstContenderReceivedGrant).isTrue();
+ assertThat(secondContenderReceivedGrant).isTrue();
+ }
}
- };
+ }
}
@Test
void testDelayedGrantCallAfterContenderBeingDeregisteredAgain() throws Exception {
- new Context() {
- {
- runTestWithManuallyTriggeredEvents(
- executorService -> {
- // we need to close the LeaderElection to deregister the contender that
- // was already registered to the service
- closeLeaderElectionInBothContexts();
-
- grantLeadership();
- executorService.trigger();
-
- applyToBothContenderContexts(
- ctx -> {
- ctx.leaderElection =
- leaderElectionService.createLeaderElection(
- ctx.contenderID);
- final TestingContender contender =
- new TestingContender(
- ctx.address + "-different",
- ctx.leaderElection);
- contender.startLeaderElection();
-
- ctx.leaderElection.close();
-
- executorService.trigger();
- });
- });
+ final TestingLeaderElectionDriver.Factory driverFactory =
+ new TestingLeaderElectionDriver.Factory(
+ TestingLeaderElectionDriver.newNoOpBuilder());
+ final ManuallyTriggeredScheduledExecutorService leaderEventOperationExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+ try (final DefaultLeaderElectionService leaderElectionService =
+ new DefaultLeaderElectionService(
+ driverFactory,
+ fatalErrorHandlerExtension.getTestingFatalErrorHandler(),
+ leaderEventOperationExecutor)) {
+
+ final AtomicBoolean leadershipGrantForwardedToContender = new AtomicBoolean(false);
+ final LeaderContender leaderContender =
+ TestingGenericLeaderContender.newBuilder()
+ .setGrantLeadershipConsumer(
+ ignoredSessionID ->
+ leadershipGrantForwardedToContender.set(true))
+ .build();
+ try (final LeaderElection leaderElection =
+ leaderElectionService.createLeaderElection("contender_id")) {
+ leaderElection.startLeaderElection(leaderContender);
+
+ assertThat(driverFactory.getCreatedDriverCount())
+ .as(
+ "A single driver should have been created when registering the contender.")
+ .isEqualTo(1);
+ leaderElectionService.onGrantLeadership(UUID.randomUUID());
}
- };
+
+ leaderEventOperationExecutor.trigger();
+ assertThat(leadershipGrantForwardedToContender).isFalse();
+ }
}
@Test
@@ -367,25 +469,6 @@ class DefaultLeaderElectionServiceTest {
};
}
- @Test
- void testContenderRegistrationWithoutDriverBeingInstantiatedFails() throws Exception {
- try (final DefaultLeaderElectionService leaderElectionService =
- new DefaultLeaderElectionService(
- TestingLeaderElectionDriver.Factory.createFactoryWithNoOpDriver(),
- fatalErrorHandlerExtension.getTestingFatalErrorHandler())) {
- final LeaderElection leaderElection =
- leaderElectionService.createLeaderElection(createRandomContenderID());
- assertThatThrownBy(
- () ->
- new TestingContender("unused-address", leaderElection)
- .startLeaderElection())
- .isInstanceOf(IllegalStateException.class);
-
- // starting the backend because the close method expects it to be initialized
- leaderElectionService.startLeaderElectionBackend();
- }
- }
-
@Test
void testDriverShutdownFailsWithContenderStillBeingRegistered() throws Exception {
new Context() {
@@ -1125,7 +1208,6 @@ class DefaultLeaderElectionServiceTest {
final DefaultLeaderElectionService testInstance =
new DefaultLeaderElectionService(
driverFactory, fatalErrorHandlerExtension.getTestingFatalErrorHandler());
- testInstance.startLeaderElectionBackend();
final LeaderElection leaderElection =
testInstance.createLeaderElection(createRandomContenderID());
@@ -1225,8 +1307,6 @@ class DefaultLeaderElectionServiceTest {
.getTestingFatalErrorHandler(),
leaderEventOperationExecutor)) {
leaderElectionService = localLeaderElectionService;
- leaderElectionService.startLeaderElectionBackend();
- testingLeaderElectionDriver = driverFactory.assertAndGetOnlyCreatedDriver();
try (final ContenderContext localContenderContext0 =
ContenderContext.create(0, leaderElectionService);
@@ -1235,6 +1315,7 @@ class DefaultLeaderElectionServiceTest {
this.contenderContext0 = localContenderContext0;
this.contenderContext1 = localContenderContext1;
+ testingLeaderElectionDriver = driverFactory.assertAndGetOnlyCreatedDriver();
testMethod.run();
}
} finally {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderElectionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderElectionTest.java
index a2bb163c6fb..6b379182b91 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderElectionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderElectionTest.java
@@ -179,7 +179,6 @@ public class LeaderElectionTest {
new ZooKeeperLeaderElectionDriverFactory(
curatorFrameworkWrapper.asCuratorFramework());
leaderElectionService = new DefaultLeaderElectionService(driverFactory);
- leaderElectionService.startLeaderElectionBackend();
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderElectionDriver.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderElectionDriver.java
index 95f3d79f025..07610b67845 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderElectionDriver.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderElectionDriver.java
@@ -216,6 +216,10 @@ public class TestingLeaderElectionDriver implements LeaderElectionDriver {
return driver;
}
+
+ public int getCreatedDriverCount() {
+ return createdDrivers.size();
+ }
}
/** {@link Builder} for creating {@link TestingLeaderElectionDriver} instances. */
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionConnectionHandlingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionConnectionHandlingTest.java
index 205e79d7495..c9eb465d770 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionConnectionHandlingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionConnectionHandlingTest.java
@@ -139,7 +139,6 @@ class ZooKeeperLeaderElectionConnectionHandlingTest {
new DefaultLeaderElectionService(
leaderElectionDriverFactory,
testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
- leaderElectionService.startLeaderElectionBackend();
final TestingConnectionStateListener connectionStateListener =
new TestingConnectionStateListener();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java
index 09fce91d8ff..a68b132b287 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java
@@ -175,11 +175,7 @@ class ZooKeeperLeaderElectionTest {
for (int i = 0; i < num; i++) {
final LeaderElectionDriverFactory driverFactory =
new ZooKeeperLeaderElectionDriverFactory(createZooKeeperClient());
- leaderElectionService[i] =
- new DefaultLeaderElectionService(
- driverFactory,
- testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
- leaderElectionService[i].startLeaderElectionBackend();
+ leaderElectionService[i] = new DefaultLeaderElectionService(driverFactory);
leaderElections[i] = leaderElectionService[i].createLeaderElection(CONTENDER_ID);
contenders[i] = new TestingContender(createAddress(i), leaderElections[i]);
@@ -279,11 +275,7 @@ class ZooKeeperLeaderElectionTest {
for (int i = 0; i < num; i++) {
final LeaderElectionDriverFactory driverFactory =
new ZooKeeperLeaderElectionDriverFactory(createZooKeeperClient());
- leaderElectionService[i] =
- new DefaultLeaderElectionService(
- driverFactory,
- testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
- leaderElectionService[i].startLeaderElectionBackend();
+ leaderElectionService[i] = new DefaultLeaderElectionService(driverFactory);
leaderElections[i] = leaderElectionService[i].createLeaderElection(CONTENDER_ID);
contenders[i] =
new TestingContender(LEADER_ADDRESS + "_" + i + "_0", leaderElections[i]);
@@ -317,11 +309,7 @@ class ZooKeeperLeaderElectionTest {
// create new leader election service which takes part in the leader election
final LeaderElectionDriverFactory driverFactory =
new ZooKeeperLeaderElectionDriverFactory(createZooKeeperClient());
- leaderElectionService[index] =
- new DefaultLeaderElectionService(
- driverFactory,
- testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
- leaderElectionService[index].startLeaderElectionBackend();
+ leaderElectionService[index] = new DefaultLeaderElectionService(driverFactory);
leaderElections[index] =
leaderElectionService[index].createLeaderElection(CONTENDER_ID);