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/06 18:39:33 UTC
[flink] 01/02: [FLINK-32419][runtime] Deletes obsolete/unused leader election related classes
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 716aa11ad516e1619f979ee635cb53f2dbe28e1b
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Fri Jun 23 10:29:33 2023 +0200
[FLINK-32419][runtime] Deletes obsolete/unused leader election related classes
- Deletes LeaderElectionDriver- and LeaderElectionEventHandler-related classes
- Deletes old zk-related classes/methods/interfaces
- Deletes old k8s-related classes
- Deletes MultipleComponentLeaderElection-related classes and interfaces
---
.../KubernetesLeaderElectionDriver.java | 260 ------------
.../KubernetesLeaderElectionDriverFactory.java | 64 ---
.../KubernetesLeaderRetrievalDriver.java | 14 +-
.../KubernetesLeaderRetrievalDriverFactory.java | 62 ---
...leComponentLeaderElectionHaServicesFactory.java | 42 --
.../KubernetesLeaderElectionDriverTest.java | 2 +-
...sMultipleComponentLeaderElectionDriverTest.java | 237 -----------
.../DefaultLeaderElectionService.java | 40 +-
...aultMultipleComponentLeaderElectionService.java | 258 ------------
.../leaderelection/LeaderElectionDriver.java | 52 ---
.../LeaderElectionDriverFactory.java | 39 --
.../leaderelection/LeaderElectionEventHandler.java | 72 ----
.../leaderelection/LeaderElectionException.java | 6 +-
...ltipleComponentLeaderElectionDriverAdapter.java | 111 -----
...omponentLeaderElectionDriverAdapterFactory.java | 52 ---
.../MultipleComponentLeaderElectionService.java | 78 ----
.../ZooKeeperLeaderElectionDriver.java | 239 -----------
.../ZooKeeperLeaderElectionDriverFactory.java | 48 ---
.../leaderretrieval/LeaderRetrievalDriver.java | 3 +-
.../ZooKeeperLeaderRetrievalDriver.java | 10 +-
.../DefaultLeaderElectionServiceTest.java | 52 +--
...MultipleComponentLeaderElectionServiceTest.java | 448 ---------------------
.../runtime/leaderelection/TestingLeaderBase.java | 2 +-
.../ZooKeeperLeaderElectionTest.java | 8 +-
24 files changed, 32 insertions(+), 2167 deletions(-)
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriver.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriver.java
deleted file mode 100644
index ec0dfe30e46..00000000000
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriver.java
+++ /dev/null
@@ -1,260 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.kubernetes.highavailability;
-
-import org.apache.flink.kubernetes.configuration.KubernetesLeaderElectionConfiguration;
-import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
-import org.apache.flink.kubernetes.kubeclient.KubernetesConfigMapSharedWatcher;
-import org.apache.flink.kubernetes.kubeclient.KubernetesSharedWatcher.Watch;
-import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
-import org.apache.flink.kubernetes.kubeclient.resources.KubernetesException;
-import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
-import org.apache.flink.kubernetes.utils.KubernetesUtils;
-import org.apache.flink.runtime.leaderelection.LeaderElectionDriver;
-import org.apache.flink.runtime.leaderelection.LeaderElectionEventHandler;
-import org.apache.flink.runtime.leaderelection.LeaderElectionException;
-import org.apache.flink.runtime.leaderelection.LeaderInformation;
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.UUID;
-import java.util.concurrent.ExecutorService;
-
-import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
-import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
-import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
-import static org.apache.flink.kubernetes.utils.KubernetesUtils.getLeaderInformationFromConfigMap;
-import static org.apache.flink.kubernetes.utils.KubernetesUtils.getOnlyConfigMap;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * {@link LeaderElectionDriver} implementation for Kubernetes. The active leader is elected using
- * Kubernetes. The current leader's address as well as its leader session ID is published via
- * Kubernetes ConfigMap. Note that the contending lock and leader storage are using the same
- * ConfigMap. And every component(e.g. ResourceManager, Dispatcher, RestEndpoint, JobManager for
- * each job) will have a separate ConfigMap.
- *
- * @deprecated in favour of {@link KubernetesMultipleComponentLeaderElectionDriver}
- */
-@Deprecated
-public class KubernetesLeaderElectionDriver implements LeaderElectionDriver {
-
- private static final Logger LOG = LoggerFactory.getLogger(KubernetesLeaderElectionDriver.class);
-
- private final FlinkKubeClient kubeClient;
-
- private final String configMapName;
-
- private final String lockIdentity;
-
- private final KubernetesLeaderElector leaderElector;
-
- // Labels will be used to clean up the ha related ConfigMaps.
- private final Map<String, String> configMapLabels;
-
- private final LeaderElectionEventHandler leaderElectionEventHandler;
-
- private final FatalErrorHandler fatalErrorHandler;
-
- private volatile boolean running;
-
- private final Watch kubernetesWatch;
-
- public KubernetesLeaderElectionDriver(
- FlinkKubeClient kubeClient,
- KubernetesConfigMapSharedWatcher configMapSharedWatcher,
- ExecutorService watchExecutorService,
- KubernetesLeaderElectionConfiguration leaderConfig,
- LeaderElectionEventHandler leaderElectionEventHandler,
- FatalErrorHandler fatalErrorHandler) {
-
- this.kubeClient = checkNotNull(kubeClient, "Kubernetes client");
- checkNotNull(leaderConfig, "Leader election configuration");
- this.leaderElectionEventHandler =
- checkNotNull(leaderElectionEventHandler, "LeaderElectionEventHandler");
- this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
-
- this.configMapName = leaderConfig.getConfigMapName();
- this.lockIdentity = leaderConfig.getLockIdentity();
- this.leaderElector =
- kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
- this.configMapLabels =
- KubernetesUtils.getConfigMapLabels(
- leaderConfig.getClusterId(), LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
-
- running = true;
- leaderElector.run();
- kubernetesWatch =
- checkNotNull(configMapSharedWatcher, "ConfigMap Shared Informer")
- .watch(
- configMapName,
- new ConfigMapCallbackHandlerImpl(),
- watchExecutorService);
- }
-
- @Override
- public void close() {
- if (!running) {
- return;
- }
- running = false;
-
- LOG.info("Closing {}.", this);
- leaderElector.stop();
-
- kubernetesWatch.close();
- }
-
- @Override
- public void writeLeaderInformation(LeaderInformation leaderInformation) {
- assert (running);
- final UUID confirmedLeaderSessionID = leaderInformation.getLeaderSessionID();
- final String confirmedLeaderAddress = leaderInformation.getLeaderAddress();
- try {
- kubeClient
- .checkAndUpdateConfigMap(
- configMapName,
- configMap -> {
- if (KubernetesLeaderElector.hasLeadership(
- configMap, lockIdentity)) {
- // Get the updated ConfigMap with new leader information
- if (confirmedLeaderAddress == null) {
- configMap.getData().remove(LEADER_ADDRESS_KEY);
- } else {
- configMap
- .getData()
- .put(LEADER_ADDRESS_KEY, confirmedLeaderAddress);
- }
- if (confirmedLeaderSessionID == null) {
- configMap.getData().remove(LEADER_SESSION_ID_KEY);
- } else {
- configMap
- .getData()
- .put(
- LEADER_SESSION_ID_KEY,
- confirmedLeaderSessionID.toString());
- }
- configMap.getLabels().putAll(configMapLabels);
- return Optional.of(configMap);
- }
- return Optional.empty();
- })
- .get();
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Successfully wrote leader information: Leader={}, session ID={}.",
- confirmedLeaderAddress,
- confirmedLeaderSessionID);
- }
- } catch (Exception e) {
- fatalErrorHandler.onFatalError(
- new KubernetesException(
- "Could not write leader information since ConfigMap "
- + configMapName
- + " does not exist.",
- e));
- }
- }
-
- @Override
- public boolean hasLeadership() {
- assert (running);
- final Optional<KubernetesConfigMap> configMapOpt = kubeClient.getConfigMap(configMapName);
- if (configMapOpt.isPresent()) {
- return KubernetesLeaderElector.hasLeadership(configMapOpt.get(), lockIdentity);
- } else {
- fatalErrorHandler.onFatalError(
- new KubernetesException(
- "ConfigMap " + configMapName + " does not exist.", null));
- return false;
- }
- }
-
- private class LeaderCallbackHandlerImpl extends KubernetesLeaderElector.LeaderCallbackHandler {
-
- @Override
- public void isLeader() {
- leaderElectionEventHandler.onGrantLeadership(UUID.randomUUID());
- }
-
- @Override
- public void notLeader() {
- leaderElectionEventHandler.onRevokeLeadership();
- // Continue to contend the leader
- leaderElector.run();
- }
- }
-
- private class ConfigMapCallbackHandlerImpl
- implements FlinkKubeClient.WatchCallbackHandler<KubernetesConfigMap> {
- @Override
- public void onAdded(List<KubernetesConfigMap> configMaps) {
- // noop
- }
-
- @Override
- public void onModified(List<KubernetesConfigMap> configMaps) {
- // We should only receive events for the watched ConfigMap
- final KubernetesConfigMap configMap = getOnlyConfigMap(configMaps, configMapName);
-
- if (KubernetesLeaderElector.hasLeadership(configMap, lockIdentity)) {
- leaderElectionEventHandler.onLeaderInformationChange(
- getLeaderInformationFromConfigMap(configMap));
- }
- }
-
- @Override
- public void onDeleted(List<KubernetesConfigMap> configMaps) {
- final KubernetesConfigMap configMap = getOnlyConfigMap(configMaps, configMapName);
- // The ConfigMap is deleted externally.
- if (KubernetesLeaderElector.hasLeadership(configMap, lockIdentity)) {
- fatalErrorHandler.onFatalError(
- new LeaderElectionException(
- "ConfigMap " + configMapName + " is deleted externally"));
- }
- }
-
- @Override
- public void onError(List<KubernetesConfigMap> configMaps) {
- final KubernetesConfigMap configMap = getOnlyConfigMap(configMaps, configMapName);
- if (KubernetesLeaderElector.hasLeadership(configMap, lockIdentity)) {
- fatalErrorHandler.onFatalError(
- new LeaderElectionException(
- "Error while watching the ConfigMap " + configMapName));
- }
- }
-
- @Override
- public void handleError(Throwable throwable) {
- fatalErrorHandler.onFatalError(
- new LeaderElectionException(
- "Error while watching the ConfigMap " + configMapName, throwable));
- }
- }
-
- @Override
- public String toString() {
- return "KubernetesLeaderElectionDriver{configMapName='" + configMapName + "'}";
- }
-}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriverFactory.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriverFactory.java
deleted file mode 100644
index bb4dce472a3..00000000000
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriverFactory.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.kubernetes.highavailability;
-
-import org.apache.flink.kubernetes.configuration.KubernetesLeaderElectionConfiguration;
-import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
-import org.apache.flink.kubernetes.kubeclient.KubernetesConfigMapSharedWatcher;
-import org.apache.flink.runtime.leaderelection.LeaderElectionDriverFactory;
-import org.apache.flink.runtime.leaderelection.LeaderElectionEventHandler;
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
-
-import java.util.concurrent.ExecutorService;
-
-/**
- * {@link LeaderElectionDriverFactory} implementation for Kubernetes.
- *
- * @deprecated in favour of {@link KubernetesMultipleComponentLeaderElectionDriverFactory}
- */
-public class KubernetesLeaderElectionDriverFactory implements LeaderElectionDriverFactory {
-
- private final FlinkKubeClient kubeClient;
- private final KubernetesConfigMapSharedWatcher configMapSharedWatcher;
- private final ExecutorService watchExecutorService;
- private final KubernetesLeaderElectionConfiguration leaderConfig;
-
- public KubernetesLeaderElectionDriverFactory(
- FlinkKubeClient kubeClient,
- KubernetesConfigMapSharedWatcher configMapSharedWatcher,
- ExecutorService watchExecutorService,
- KubernetesLeaderElectionConfiguration leaderConfig) {
- this.kubeClient = kubeClient;
- this.configMapSharedWatcher = configMapSharedWatcher;
- this.watchExecutorService = watchExecutorService;
- this.leaderConfig = leaderConfig;
- }
-
- @Override
- public KubernetesLeaderElectionDriver createLeaderElectionDriver(
- LeaderElectionEventHandler leaderEventHandler, FatalErrorHandler fatalErrorHandler) {
- return new KubernetesLeaderElectionDriver(
- kubeClient,
- configMapSharedWatcher,
- watchExecutorService,
- leaderConfig,
- leaderEventHandler,
- fatalErrorHandler);
- }
-}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriver.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriver.java
index 035057c7441..88851df62b0 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriver.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriver.java
@@ -39,10 +39,10 @@ import static org.apache.flink.kubernetes.utils.KubernetesUtils.getOnlyConfigMap
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
- * The counterpart to the {@link KubernetesLeaderElectionDriver}. {@link LeaderRetrievalDriver}
- * implementation for Kubernetes. It retrieves the current leader which has been elected by the
- * {@link KubernetesLeaderElectionDriver}. The leader address as well as the current leader session
- * ID is retrieved from Kubernetes ConfigMap.
+ * The counterpart to the {@link KubernetesMultipleComponentLeaderElectionDriver}. {@link
+ * LeaderRetrievalDriver} implementation for Kubernetes. It retrieves the current leader which has
+ * been elected by the {@link KubernetesMultipleComponentLeaderElectionDriver}. The leader address
+ * as well as the current leader session ID is retrieved from Kubernetes ConfigMap.
*/
public class KubernetesLeaderRetrievalDriver implements LeaderRetrievalDriver {
@@ -98,9 +98,9 @@ public class KubernetesLeaderRetrievalDriver implements LeaderRetrievalDriver {
@Override
public void onAdded(List<KubernetesConfigMap> configMaps) {
- // The ConfigMap is created by KubernetesLeaderElectionDriver with empty data. We don't
- // really need to process anything unless the retriever was started after the leader
- // election has already succeeded.
+ // The ConfigMap is created by KubernetesMultipleComponentLeaderElectionDriver with
+ // empty data. We don't really need to process anything unless the retriever was started
+ // after the leader election has already succeeded.
final KubernetesConfigMap configMap = getOnlyConfigMap(configMaps, configMapName);
final LeaderInformation leaderInformation = leaderInformationExtractor.apply(configMap);
if (!leaderInformation.isEmpty()) {
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java
deleted file mode 100644
index 408f0f6f15f..00000000000
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.kubernetes.highavailability;
-
-import org.apache.flink.kubernetes.kubeclient.KubernetesConfigMapSharedWatcher;
-import org.apache.flink.kubernetes.utils.KubernetesUtils;
-import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalDriverFactory;
-import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalEventHandler;
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
-
-import java.util.concurrent.Executor;
-
-/**
- * {@link LeaderRetrievalDriverFactory} implementation for Kubernetes.
- *
- * @deprecated in favour of {@link KubernetesMultipleComponentLeaderRetrievalDriverFactory}
- */
-@Deprecated
-public class KubernetesLeaderRetrievalDriverFactory implements LeaderRetrievalDriverFactory {
-
- private final KubernetesConfigMapSharedWatcher configMapSharedWatcher;
- private final Executor watchExecutor;
-
- private final String configMapName;
-
- public KubernetesLeaderRetrievalDriverFactory(
- KubernetesConfigMapSharedWatcher configMapSharedWatcher,
- Executor watchExecutor,
- String configMapName) {
- this.configMapSharedWatcher = configMapSharedWatcher;
- this.watchExecutor = watchExecutor;
- this.configMapName = configMapName;
- }
-
- @Override
- public KubernetesLeaderRetrievalDriver createLeaderRetrievalDriver(
- LeaderRetrievalEventHandler leaderEventHandler, FatalErrorHandler fatalErrorHandler) {
- return new KubernetesLeaderRetrievalDriver(
- configMapSharedWatcher,
- watchExecutor,
- configMapName,
- leaderEventHandler,
- KubernetesUtils::getLeaderInformationFromConfigMap,
- fatalErrorHandler);
- }
-}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionHaServicesFactory.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionHaServicesFactory.java
deleted file mode 100644
index bf3d8eb45bf..00000000000
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionHaServicesFactory.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.kubernetes.highavailability;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.kubernetes.kubeclient.FlinkKubeClientFactory;
-import org.apache.flink.runtime.blob.BlobUtils;
-import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
-import org.apache.flink.runtime.highavailability.HighAvailabilityServicesFactory;
-
-import java.util.concurrent.Executor;
-
-/** Factory for {@link KubernetesMultipleComponentLeaderElectionHaServices}. */
-public class KubernetesMultipleComponentLeaderElectionHaServicesFactory
- implements HighAvailabilityServicesFactory {
- @Override
- public HighAvailabilityServices createHAServices(Configuration configuration, Executor executor)
- throws Exception {
- return new KubernetesMultipleComponentLeaderElectionHaServices(
- FlinkKubeClientFactory.getInstance()
- .fromConfiguration(configuration, "kubernetes-ha-services"),
- executor,
- configuration,
- BlobUtils.createBlobStoreFromConfig(configuration));
- }
-}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriverTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriverTest.java
index 27e2d9d0be7..a4d676a264c 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriverTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriverTest.java
@@ -39,7 +39,7 @@ import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests for the {@link KubernetesLeaderElectionDriver}. */
+/** Tests for the {@link KubernetesMultipleComponentLeaderElectionDriver}. */
class KubernetesLeaderElectionDriverTest extends KubernetesHighAvailabilityTestBase {
@Test
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionDriverTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionDriverTest.java
deleted file mode 100644
index 3f2edb8f658..00000000000
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionDriverTest.java
+++ /dev/null
@@ -1,237 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.kubernetes.highavailability;
-
-import org.apache.flink.kubernetes.configuration.KubernetesLeaderElectionConfiguration;
-import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
-import org.apache.flink.kubernetes.kubeclient.KubernetesConfigMapSharedWatcher;
-import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
-import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
-import org.apache.flink.runtime.leaderelection.LeaderElectionEvent;
-import org.apache.flink.runtime.leaderelection.LeaderInformation;
-import org.apache.flink.runtime.leaderelection.LeaderInformationRegister;
-import org.apache.flink.runtime.leaderelection.TestingLeaderElectionListener;
-import org.apache.flink.runtime.leaderelection.TestingListener;
-import org.apache.flink.runtime.leaderretrieval.DefaultLeaderRetrievalService;
-import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
-import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorExtension;
-import org.apache.flink.util.function.RunnableWithException;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
-import java.util.Collections;
-import java.util.UUID;
-import java.util.concurrent.ScheduledExecutorService;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for the {@link KubernetesMultipleComponentLeaderElectionDriver}. */
-class KubernetesMultipleComponentLeaderElectionDriverTest {
-
- private static final String CLUSTER_ID = "test-cluster";
- private static final String LEADER_CONFIGMAP_NAME = "leader-configmap-name";
- private static final String LOCK_IDENTITY = "lock-identity";
-
- @RegisterExtension
- private final TestingFatalErrorHandlerExtension testingFatalErrorHandlerExtension =
- new TestingFatalErrorHandlerExtension();
-
- @RegisterExtension
- private static final TestExecutorExtension<ScheduledExecutorService> testExecutorExtension =
- TestingUtils.defaultExecutorExtension();
-
- @Test
- void testElectionDriverGainsLeadership() throws Exception {
- new TestFixture() {
- {
- runTest(
- () -> {
- leaderCallbackGrantLeadership();
- leaderElectionListener.await(LeaderElectionEvent.IsLeaderEvent.class);
- });
- }
- };
- }
-
- @Test
- void testElectionDriverLosesLeadership() throws Exception {
- new TestFixture() {
- {
- runTest(
- () -> {
- leaderCallbackGrantLeadership();
- leaderElectionListener.await(LeaderElectionEvent.IsLeaderEvent.class);
- getLeaderCallback().notLeader();
- leaderElectionListener.await(LeaderElectionEvent.NotLeaderEvent.class);
- });
- }
- };
- }
-
- @Test
- void testPublishLeaderInformation() throws Exception {
- new TestFixture() {
- {
- runTest(
- () -> {
- leaderCallbackGrantLeadership();
- leaderElectionListener.await(LeaderElectionEvent.IsLeaderEvent.class);
- final LeaderInformation leaderInformation =
- LeaderInformation.known(UUID.randomUUID(), "localhost");
- final String componentId = "componentId";
-
- final DefaultLeaderRetrievalService leaderRetrievalService =
- new DefaultLeaderRetrievalService(
- new KubernetesMultipleComponentLeaderRetrievalDriverFactory(
- getConfigMapSharedWatcher(),
- testExecutorExtension.getExecutor(),
- LEADER_CONFIGMAP_NAME,
- componentId));
-
- final TestingListener leaderRetrievalListener = new TestingListener();
- leaderRetrievalService.start(leaderRetrievalListener);
-
- leaderElectionDriver.publishLeaderInformation(
- componentId, leaderInformation);
-
- notifyLeaderRetrievalWatchOnModifiedConfigMap();
-
- leaderRetrievalListener.waitForNewLeader();
- assertThat(leaderRetrievalListener.getLeader())
- .isEqualTo(leaderInformation);
- });
- }
- };
- }
-
- @Test
- void testLeaderInformationChangeNotifiesListener() throws Exception {
- new TestFixture() {
- {
- runTest(
- () -> {
- leaderCallbackGrantLeadership();
- final String componentA = "componentA";
- final LeaderInformation leaderInformationA =
- LeaderInformation.known(UUID.randomUUID(), "localhost");
- final String componentB = "componentB";
- final LeaderInformation leaderInformationB =
- LeaderInformation.known(UUID.randomUUID(), "localhost");
- leaderElectionDriver.publishLeaderInformation(
- componentA, leaderInformationA);
- leaderElectionDriver.publishLeaderInformation(
- componentB, leaderInformationB);
-
- notifyLeaderElectionWatchOnModifiedConfigMap();
-
- final LeaderElectionEvent.AllKnownLeaderInformationEvent
- allKnownLeaderInformationEvent =
- leaderElectionListener.await(
- LeaderElectionEvent
- .AllKnownLeaderInformationEvent.class);
-
- final LeaderInformationRegister actualLeaderInformationRegister =
- allKnownLeaderInformationEvent.getLeaderInformationRegister();
-
- assertThat(actualLeaderInformationRegister.getRegisteredContenderIDs())
- .containsExactlyInAnyOrder(componentA, componentB);
- assertThat(actualLeaderInformationRegister.forContenderID(componentA))
- .hasValue(leaderInformationA);
- assertThat(actualLeaderInformationRegister.forContenderID(componentB))
- .hasValue(leaderInformationB);
- });
- }
- };
- }
-
- /** Test fixture for the {@link KubernetesMultipleComponentLeaderElectionDriverTest}. */
- protected class TestFixture {
- private final KubernetesTestFixture kubernetesTestFixture;
- final TestingLeaderElectionListener leaderElectionListener;
- final KubernetesMultipleComponentLeaderElectionDriver leaderElectionDriver;
-
- TestFixture() {
- kubernetesTestFixture =
- new KubernetesTestFixture(CLUSTER_ID, LEADER_CONFIGMAP_NAME, LOCK_IDENTITY);
- leaderElectionListener = new TestingLeaderElectionListener();
- leaderElectionDriver = createLeaderElectionDriver();
- }
-
- private KubernetesMultipleComponentLeaderElectionDriver createLeaderElectionDriver() {
- final KubernetesLeaderElectionConfiguration leaderElectionConfiguration =
- new KubernetesLeaderElectionConfiguration(
- LEADER_CONFIGMAP_NAME,
- LOCK_IDENTITY,
- kubernetesTestFixture.getConfiguration());
- return new KubernetesMultipleComponentLeaderElectionDriver(
- leaderElectionConfiguration,
- kubernetesTestFixture.getFlinkKubeClient(),
- leaderElectionListener,
- kubernetesTestFixture.getConfigMapSharedWatcher(),
- testExecutorExtension.getExecutor(),
- testingFatalErrorHandlerExtension.getTestingFatalErrorHandler());
- }
-
- void leaderCallbackGrantLeadership() throws Exception {
- kubernetesTestFixture.leaderCallbackGrantLeadership();
- }
-
- KubernetesLeaderElector.LeaderCallbackHandler getLeaderCallback() throws Exception {
- return kubernetesTestFixture.getLeaderCallback();
- }
-
- FlinkKubeClient getFlinkKubeClient() {
- return kubernetesTestFixture.getFlinkKubeClient();
- }
-
- KubernetesConfigMapSharedWatcher getConfigMapSharedWatcher() {
- return kubernetesTestFixture.getConfigMapSharedWatcher();
- }
-
- FlinkKubeClient.WatchCallbackHandler<KubernetesConfigMap>
- getLeaderRetrievalConfigMapCallback() throws Exception {
- return kubernetesTestFixture.getLeaderRetrievalConfigMapCallback();
- }
-
- void notifyLeaderRetrievalWatchOnModifiedConfigMap() throws Exception {
- kubernetesTestFixture
- .getLeaderRetrievalConfigMapCallback()
- .onModified(
- Collections.singletonList(kubernetesTestFixture.getLeaderConfigMap()));
- }
-
- void notifyLeaderElectionWatchOnModifiedConfigMap() throws Exception {
- kubernetesTestFixture
- .getLeaderElectionConfigMapCallback()
- .onModified(
- Collections.singletonList(kubernetesTestFixture.getLeaderConfigMap()));
- }
-
- void runTest(RunnableWithException testMethod) throws Exception {
- try {
- testMethod.run();
- } finally {
- leaderElectionDriver.close();
- kubernetesTestFixture.close();
- }
- }
- }
-}
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 030beaef212..6d73db5fd2e 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
@@ -42,15 +42,13 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* Default implementation for leader election service. Composed with different {@link
- * LeaderElectionDriver}, we could perform a leader election for the contender, and then persist the
- * leader information to various storage.
+ * MultipleComponentLeaderElectionDriver}, we could perform a leader election for the contender, and
+ * then persist the leader information to various storage.
*
* <p>{@code DefaultLeaderElectionService} handles a single {@link LeaderContender}.
*/
public class DefaultLeaderElectionService extends AbstractLeaderElectionService
- implements LeaderElectionEventHandler,
- MultipleComponentLeaderElectionDriver.Listener,
- AutoCloseable {
+ implements MultipleComponentLeaderElectionDriver.Listener, AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(DefaultLeaderElectionService.class);
@@ -69,10 +67,10 @@ public class DefaultLeaderElectionService extends AbstractLeaderElectionService
private final Map<String, LeaderContender> leaderContenderRegistry = new HashMap<>();
/**
- * Saves the session ID which was issued by the {@link LeaderElectionDriver} if and only if the
- * leadership is acquired by this service. {@code issuedLeaderSessionID} being {@code null}
- * indicates that this service isn't the leader right now (i.e. {@link
- * #onGrantLeadership(UUID)}) wasn't called, yet (independently of what {@code
+ * Saves the session ID which was issued by the {@link MultipleComponentLeaderElectionDriver} if
+ * and only if the leadership is acquired by this service. {@code issuedLeaderSessionID} being
+ * {@code null} indicates that this service isn't the leader right now (i.e. {@link
+ * #isLeader(UUID)}) wasn't called, yet (independently of what {@code
* leaderElectionDriver#hasLeadership()} returns).
*/
@GuardedBy("lock")
@@ -379,11 +377,6 @@ public class DefaultLeaderElectionService extends AbstractLeaderElectionService
}
}
- @Override
- public void onGrantLeadership(UUID newLeaderSessionId) {
- isLeader(newLeaderSessionId);
- }
-
@GuardedBy("lock")
private void onGrantLeadershipInternal(UUID newLeaderSessionId) {
Preconditions.checkNotNull(newLeaderSessionId);
@@ -430,11 +423,6 @@ public class DefaultLeaderElectionService extends AbstractLeaderElectionService
leaderContenderRegistry.get(contenderID).grantLeadership(issuedLeaderSessionID);
}
- @Override
- public void onRevokeLeadership() {
- notLeader();
- }
-
@GuardedBy("lock")
private void onRevokeLeadershipInternal() {
// TODO: FLINK-31814 covers adding this Precondition
@@ -477,20 +465,6 @@ public class DefaultLeaderElectionService extends AbstractLeaderElectionService
leaderContender.revokeLeadership();
}
- @Override
- public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- if (leaderContenderRegistry.size() > 1) {
- throw new UnsupportedOperationException(
- "The only way this method works is when there's only a single contender registered.");
- }
-
- leaderContenderRegistry
- .keySet()
- .forEach(
- contenderID ->
- notifyLeaderInformationChange(contenderID, leaderInformation));
- }
-
@GuardedBy("lock")
private void notifyLeaderInformationChangeInternal(
String contenderID,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultMultipleComponentLeaderElectionService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultMultipleComponentLeaderElectionService.java
deleted file mode 100644
index 2775b1c5ce9..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultMultipleComponentLeaderElectionService.java
+++ /dev/null
@@ -1,258 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.leaderelection;
-
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
-import org.apache.flink.util.ExecutorUtils;
-import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.concurrent.ExecutorThreadFactory;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nullable;
-import javax.annotation.concurrent.GuardedBy;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
-
-/**
- * Default implementation of a {@link MultipleComponentLeaderElectionService} that allows to
- * register multiple {@link LeaderElectionEventHandler}.
- */
-public class DefaultMultipleComponentLeaderElectionService
- implements MultipleComponentLeaderElectionService,
- MultipleComponentLeaderElectionDriver.Listener {
- private static final Logger LOG =
- LoggerFactory.getLogger(DefaultMultipleComponentLeaderElectionService.class);
-
- private final Object lock = new Object();
-
- private final MultipleComponentLeaderElectionDriver multipleComponentLeaderElectionDriver;
-
- @GuardedBy("lock")
- private final ExecutorService leadershipOperationExecutor;
-
- @GuardedBy("lock")
- private final Map<String, LeaderElectionEventHandler> leaderElectionEventHandlers;
-
- @GuardedBy("lock")
- private boolean running = true;
-
- @Nullable
- @GuardedBy("lock")
- private UUID currentLeaderSessionId = null;
-
- @VisibleForTesting
- DefaultMultipleComponentLeaderElectionService(
- FatalErrorHandler fatalErrorHandler,
- MultipleComponentLeaderElectionDriverFactory
- multipleComponentLeaderElectionDriverFactory,
- ExecutorService leadershipOperationExecutor)
- throws Exception {
- this.leadershipOperationExecutor = Preconditions.checkNotNull(leadershipOperationExecutor);
-
- leaderElectionEventHandlers = new HashMap<>();
-
- multipleComponentLeaderElectionDriver =
- multipleComponentLeaderElectionDriverFactory.create(this, fatalErrorHandler);
- }
-
- public DefaultMultipleComponentLeaderElectionService(
- FatalErrorHandler fatalErrorHandler,
- MultipleComponentLeaderElectionDriverFactory
- multipleComponentLeaderElectionDriverFactory)
- throws Exception {
- this(
- fatalErrorHandler,
- multipleComponentLeaderElectionDriverFactory,
- Executors.newSingleThreadExecutor(
- new ExecutorThreadFactory("leadershipOperationExecutor")));
- }
-
- @Override
- public void close() throws Exception {
- synchronized (lock) {
- if (!running) {
- return;
- }
- running = false;
-
- LOG.info("Closing {}.", this.getClass().getSimpleName());
-
- ExecutorUtils.gracefulShutdown(10L, TimeUnit.SECONDS, leadershipOperationExecutor);
-
- multipleComponentLeaderElectionDriver.close();
- }
- }
-
- @Override
- public MultipleComponentLeaderElectionDriverFactory createDriverFactory(String componentId) {
- return new MultipleComponentLeaderElectionDriverAdapterFactory(componentId, this);
- }
-
- @Override
- public void publishLeaderInformation(String componentId, LeaderInformation leaderInformation) {
- multipleComponentLeaderElectionDriver.publishLeaderInformation(
- componentId, leaderInformation);
- }
-
- @Override
- public void registerLeaderElectionEventHandler(
- String componentId, LeaderElectionEventHandler leaderElectionEventHandler) {
-
- synchronized (lock) {
- Preconditions.checkArgument(
- !leaderElectionEventHandlers.containsKey(componentId),
- "Do not support duplicate LeaderElectionEventHandler registration under %s",
- componentId);
- leaderElectionEventHandlers.put(componentId, leaderElectionEventHandler);
-
- if (currentLeaderSessionId != null) {
- final UUID leaderSessionId = currentLeaderSessionId;
- leadershipOperationExecutor.execute(
- () -> leaderElectionEventHandler.onGrantLeadership(leaderSessionId));
- }
- }
- }
-
- @Override
- public void unregisterLeaderElectionEventHandler(String componentId) throws Exception {
- final LeaderElectionEventHandler unregisteredLeaderElectionEventHandler;
- synchronized (lock) {
- unregisteredLeaderElectionEventHandler =
- leaderElectionEventHandlers.remove(componentId);
-
- if (unregisteredLeaderElectionEventHandler != null) {
- leadershipOperationExecutor.execute(
- unregisteredLeaderElectionEventHandler::onRevokeLeadership);
- } else {
- LOG.debug(
- "Could not find leader election event handler for componentId {}. Ignoring the unregister call.",
- componentId);
- }
- }
-
- multipleComponentLeaderElectionDriver.deleteLeaderInformation(componentId);
- }
-
- @Override
- public boolean hasLeadership(String componentId) {
- synchronized (lock) {
- Preconditions.checkState(running);
-
- return leaderElectionEventHandlers.containsKey(componentId)
- && multipleComponentLeaderElectionDriver.hasLeadership();
- }
- }
-
- @Override
- public void isLeader(UUID leaderSessionID) {
- synchronized (lock) {
- if (!running) {
- return;
- }
-
- Preconditions.checkState(
- currentLeaderSessionId == null,
- "notLeader() wasn't called by the LeaderElection backend before assigning leadership to this LeaderElectionService.");
- currentLeaderSessionId = leaderSessionID;
-
- forEachLeaderElectionEventHandler(
- leaderElectionEventHandler ->
- leaderElectionEventHandler.onGrantLeadership(leaderSessionID));
- }
- }
-
- @Override
- public void notLeader() {
- synchronized (lock) {
- if (!running) {
- return;
- }
-
- currentLeaderSessionId = null;
-
- forEachLeaderElectionEventHandler(LeaderElectionEventHandler::onRevokeLeadership);
- }
- }
-
- @GuardedBy("lock")
- private void forEachLeaderElectionEventHandler(
- Consumer<? super LeaderElectionEventHandler> action) {
-
- for (LeaderElectionEventHandler leaderElectionEventHandler :
- leaderElectionEventHandlers.values()) {
- leadershipOperationExecutor.execute(() -> action.accept(leaderElectionEventHandler));
- }
- }
-
- @Override
- public void notifyLeaderInformationChange(
- String componentId, LeaderInformation leaderInformation) {
- synchronized (lock) {
- if (!running) {
- return;
- }
-
- final LeaderElectionEventHandler leaderElectionEventHandler =
- leaderElectionEventHandlers.get(componentId);
-
- if (leaderElectionEventHandler != null) {
- sendLeaderInformationChange(leaderElectionEventHandler, leaderInformation);
- }
- }
- }
-
- @GuardedBy("lock")
- private void sendLeaderInformationChange(
- LeaderElectionEventHandler leaderElectionEventHandler,
- LeaderInformation leaderInformation) {
- leadershipOperationExecutor.execute(
- () -> leaderElectionEventHandler.onLeaderInformationChange(leaderInformation));
- }
-
- @Override
- public void notifyAllKnownLeaderInformation(
- LeaderInformationRegister leaderInformationRegister) {
- synchronized (lock) {
- if (!running) {
- return;
- }
-
- for (Map.Entry<String, LeaderElectionEventHandler>
- leaderNameLeaderElectionEventHandlerPair :
- leaderElectionEventHandlers.entrySet()) {
- final String leaderName = leaderNameLeaderElectionEventHandlerPair.getKey();
- final LeaderInformation leaderInformation =
- leaderInformationRegister
- .forContenderID(leaderName)
- .orElse(LeaderInformation.empty());
- sendLeaderInformationChange(
- leaderNameLeaderElectionEventHandlerPair.getValue(), leaderInformation);
- }
- }
- }
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionDriver.java
deleted file mode 100644
index be16ca8a5a5..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionDriver.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.leaderelection;
-
-/**
- * A {@link LeaderElectionDriver} is responsible for performing the leader election and storing the
- * leader information. All the leader internal state is guarded by lock in {@link
- * LeaderElectionService}. Different driver implementations do not need to care about the lock. And
- * it should use {@link LeaderElectionEventHandler} if it want to respond to the leader change
- * events.
- *
- * <p><strong>Important</strong>: The {@link LeaderElectionDriver} could not guarantee that there is
- * no {@link LeaderElectionEventHandler} callbacks happen after {@link #close()}.
- */
-public interface LeaderElectionDriver extends AutoCloseable {
-
- /**
- * Write the current leader information to external persistent storage(e.g. Zookeeper,
- * Kubernetes ConfigMap). This is a blocking IO operation. The write operation takes effect only
- * when the driver still has the leadership.
- *
- * @param leaderInformation current leader information. It could be {@link
- * LeaderInformation#empty()}, which means the caller want to clear the leader information
- * on external storage. Please remember that the clear operation should only happen before a
- * new leader is elected and has written his leader information on the storage. Otherwise,
- * we may have a risk to wrongly update the storage with empty leader information.
- */
- void writeLeaderInformation(LeaderInformation leaderInformation);
-
- /**
- * Check whether the driver still have the leadership in the distributed coordination system.
- *
- * @return Return whether the driver has leadership.
- */
- boolean hasLeadership();
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionDriverFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionDriverFactory.java
deleted file mode 100644
index 07ba9acec93..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionDriverFactory.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.leaderelection;
-
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
-
-/** Factory for creating {@link LeaderElectionDriver} with different implementation. */
-public interface LeaderElectionDriverFactory {
-
- /**
- * Create a specific {@link LeaderElectionDriver} and start the necessary services. For example,
- * LeaderLatch and NodeCache in Zookeeper, KubernetesLeaderElector and ConfigMap watcher in
- * Kubernetes.
- *
- * @param leaderEventHandler handler for the leader election driver to process leader events.
- * @param fatalErrorHandler fatal error handler
- * @throws Exception when create a specific {@link LeaderElectionDriver} implementation and
- * start the necessary services.
- */
- LeaderElectionDriver createLeaderElectionDriver(
- LeaderElectionEventHandler leaderEventHandler, FatalErrorHandler fatalErrorHandler)
- throws Exception;
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionEventHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionEventHandler.java
deleted file mode 100644
index 5b25bb4566d..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionEventHandler.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.leaderelection;
-
-import javax.annotation.concurrent.NotThreadSafe;
-
-import java.util.UUID;
-
-/**
- * Interface which should be implemented to respond to leader changes in {@link
- * LeaderElectionDriver}.
- *
- * <p><strong>Important</strong>: The {@link LeaderElectionDriver} could not guarantee that there is
- * no {@link LeaderElectionEventHandler} callbacks happen after {@link
- * LeaderElectionDriver#close()}. This means that the implementor of {@link
- * LeaderElectionEventHandler} is responsible for filtering out spurious callbacks(e.g. after close
- * has been called on {@link LeaderElectionDriver}).
- *
- * <p>The order of events matters. Therefore, calling event processing functions of this interface
- * should happen in a single-thread environment.
- *
- * @deprecated {@link LeaderElectionEventHandler} will be replaced by {@link
- * MultipleComponentLeaderElectionDriver.Listener}.
- */
-@Deprecated
-@NotThreadSafe
-public interface LeaderElectionEventHandler {
-
- /**
- * Called by specific {@link LeaderElectionDriver} when the leadership is granted.
- *
- * @param newLeaderSessionId the valid leader session id
- */
- @Deprecated
- void onGrantLeadership(UUID newLeaderSessionId);
-
- /**
- * Called by specific {@link LeaderElectionDriver} when the leadership is revoked. Updating the
- * LeaderElection data at this point doesn't have any effect anymore.
- */
- @Deprecated
- void onRevokeLeadership();
-
- /**
- * Called by specific {@link LeaderElectionDriver} when the leader information is changed. Then
- * the {@link LeaderElectionService} could write the leader information again if necessary. This
- * method should only be called when {@link LeaderElectionDriver#hasLeadership()} is true.
- * Duplicated leader change events could happen, so the implementation should check whether the
- * passed leader information is really different with internal confirmed leader information.
- *
- * @param leaderInformation leader information which contains leader session id and leader
- * address.
- */
- @Deprecated
- void onLeaderInformationChange(LeaderInformation leaderInformation);
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionException.java
index 92ddceb26ad..31f5106cccb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionException.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionException.java
@@ -20,11 +20,7 @@ package org.apache.flink.runtime.leaderelection;
import org.apache.flink.util.FlinkException;
-/**
- * This exception is thrown by the {@link LeaderElectionDriver} when {@link
- * LeaderElectionDriver#writeLeaderInformation} failed or some unexpected changes to the leader
- * storage.
- */
+/** This exception is used in leader election-related scenarios. */
public class LeaderElectionException extends FlinkException {
private static final long serialVersionUID = 1L;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/MultipleComponentLeaderElectionDriverAdapter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/MultipleComponentLeaderElectionDriverAdapter.java
deleted file mode 100644
index 66f049aa6e7..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/MultipleComponentLeaderElectionDriverAdapter.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.leaderelection;
-
-import org.apache.flink.util.Preconditions;
-
-import java.util.UUID;
-
-/**
- * {@link LeaderElectionDriver} adapter that multiplexes the leader election of a component into a
- * single leader election via {@link MultipleComponentLeaderElectionService}.
- */
-final class MultipleComponentLeaderElectionDriverAdapter
- implements LeaderElectionDriver, MultipleComponentLeaderElectionDriver {
- private final String componentId;
- private final MultipleComponentLeaderElectionService multipleComponentLeaderElectionService;
-
- MultipleComponentLeaderElectionDriverAdapter(
- String componentID,
- MultipleComponentLeaderElectionService multipleComponentLeaderElectionService,
- MultipleComponentLeaderElectionDriver.Listener listener) {
- this(
- componentID,
- multipleComponentLeaderElectionService,
- // here's where the componentID of the legacy code would be translated into
- // the contenderID that will be used in the DefaultLeaderElectionService eventually
- new ListenerWrapper("unused-contender-id", listener));
- }
-
- MultipleComponentLeaderElectionDriverAdapter(
- String componentId,
- MultipleComponentLeaderElectionService multipleComponentLeaderElectionService,
- LeaderElectionEventHandler leaderElectionEventHandler) {
- this.componentId = Preconditions.checkNotNull(componentId);
- this.multipleComponentLeaderElectionService =
- Preconditions.checkNotNull(multipleComponentLeaderElectionService);
-
- multipleComponentLeaderElectionService.registerLeaderElectionEventHandler(
- this.componentId, leaderElectionEventHandler);
- }
-
- @Override
- public void writeLeaderInformation(LeaderInformation leaderInformation) {
- multipleComponentLeaderElectionService.publishLeaderInformation(
- componentId, leaderInformation);
- }
-
- @Override
- public boolean hasLeadership() {
- return multipleComponentLeaderElectionService.hasLeadership(componentId);
- }
-
- @Override
- public void publishLeaderInformation(
- String ignoredContenderID, LeaderInformation leaderInformation) {
- writeLeaderInformation(leaderInformation);
- }
-
- @Override
- public void deleteLeaderInformation(String ignoredContenderID) {
- writeLeaderInformation(LeaderInformation.empty());
- }
-
- @Override
- public void close() throws Exception {
- multipleComponentLeaderElectionService.unregisterLeaderElectionEventHandler(componentId);
- }
-
- private static class ListenerWrapper implements LeaderElectionEventHandler {
-
- private final String contenderID;
- private final MultipleComponentLeaderElectionDriver.Listener listener;
-
- public ListenerWrapper(
- String contenderID, MultipleComponentLeaderElectionDriver.Listener listener) {
- this.contenderID = contenderID;
- this.listener = listener;
- }
-
- @Override
- public void onGrantLeadership(UUID newLeaderSessionId) {
- listener.isLeader(newLeaderSessionId);
- }
-
- @Override
- public void onRevokeLeadership() {
- listener.notLeader();
- }
-
- @Override
- public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- listener.notifyLeaderInformationChange(contenderID, leaderInformation);
- }
- }
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/MultipleComponentLeaderElectionDriverAdapterFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/MultipleComponentLeaderElectionDriverAdapterFactory.java
deleted file mode 100644
index 5039b90bc52..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/MultipleComponentLeaderElectionDriverAdapterFactory.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.leaderelection;
-
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
-import org.apache.flink.util.Preconditions;
-
-/** Factory for a {@link MultipleComponentLeaderElectionDriverAdapter}. */
-final class MultipleComponentLeaderElectionDriverAdapterFactory
- implements LeaderElectionDriverFactory, MultipleComponentLeaderElectionDriverFactory {
- private final String leaderName;
- private final MultipleComponentLeaderElectionService singleLeaderElectionService;
-
- MultipleComponentLeaderElectionDriverAdapterFactory(
- String leaderName, MultipleComponentLeaderElectionService singleLeaderElectionService) {
- this.leaderName = Preconditions.checkNotNull(leaderName);
- this.singleLeaderElectionService = Preconditions.checkNotNull(singleLeaderElectionService);
- }
-
- @Override
- public LeaderElectionDriver createLeaderElectionDriver(
- LeaderElectionEventHandler leaderEventHandler, FatalErrorHandler fatalErrorHandler)
- throws Exception {
- return new MultipleComponentLeaderElectionDriverAdapter(
- leaderName, singleLeaderElectionService, leaderEventHandler);
- }
-
- @Override
- public MultipleComponentLeaderElectionDriver create(
- MultipleComponentLeaderElectionDriver.Listener leaderElectionListener,
- FatalErrorHandler fatalErrorHandler)
- throws Exception {
- return new MultipleComponentLeaderElectionDriverAdapter(
- leaderName, singleLeaderElectionService, leaderElectionListener);
- }
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/MultipleComponentLeaderElectionService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/MultipleComponentLeaderElectionService.java
deleted file mode 100644
index 1e3816538d7..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/MultipleComponentLeaderElectionService.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.leaderelection;
-
-/**
- * Leader election service that allows to register multiple {@link LeaderElectionEventHandler
- * LeaderElectionEventHandlers} that are identified by different names. For each event handler it is
- * possible to write the corresponding {@link LeaderInformation}.
- */
-public interface MultipleComponentLeaderElectionService {
-
- /**
- * Closes this service.
- *
- * @throws Exception if the service failed to close
- */
- void close() throws Exception;
-
- /**
- * Creates a {@link MultipleComponentLeaderElectionDriverFactory} for the given leader name.
- *
- * @param componentId identifying the component for which to create a leader election driver
- * factory
- * @return Leader election driver factory
- */
- MultipleComponentLeaderElectionDriverFactory createDriverFactory(String componentId);
-
- /**
- * Publishes the given leader information for the component identified by the given leader name.
- *
- * @param componentId identifying the component
- * @param leaderInformation leader information
- */
- void publishLeaderInformation(String componentId, LeaderInformation leaderInformation);
-
- /**
- * Registers a new leader election event handler under the given component id.
- *
- * @param componentId identifying the leader election event handler
- * @param leaderElectionEventHandler leader election event handler to register
- * @throws IllegalArgumentException if there is already a handler registered for the given
- * component id
- */
- void registerLeaderElectionEventHandler(
- String componentId, LeaderElectionEventHandler leaderElectionEventHandler);
-
- /**
- * Unregisters the leader election event handler with the given component id.
- *
- * @param componentId identifying the component
- * @throws Exception if the leader election event handler could not be unregistered
- */
- void unregisterLeaderElectionEventHandler(String componentId) throws Exception;
-
- /**
- * Returns whether the given component has leadership.
- *
- * @param componentId identifying the component
- * @return {@code true} if the component has leadership otherwise {@code false}
- */
- boolean hasLeadership(String componentId);
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionDriver.java
deleted file mode 100644
index cbcf02fdfa4..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionDriver.java
+++ /dev/null
@@ -1,239 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.leaderelection;
-
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
-import org.apache.flink.runtime.util.ZooKeeperUtils;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.Preconditions;
-
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework;
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.cache.ChildData;
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.cache.TreeCache;
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.leader.LeaderLatch;
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.leader.LeaderLatchListener;
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.state.ConnectionState;
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.state.ConnectionStateListener;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.UUID;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * {@link LeaderElectionDriver} implementation for Zookeeper. The leading JobManager is elected
- * using ZooKeeper. The current leader's address as well as its leader session ID is published via
- * ZooKeeper.
- *
- * @deprecated in favour of {@link ZooKeeperMultipleComponentLeaderElectionDriver}
- */
-@Deprecated
-public class ZooKeeperLeaderElectionDriver implements LeaderElectionDriver, LeaderLatchListener {
-
- private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperLeaderElectionDriver.class);
-
- /** Client to the ZooKeeper quorum. */
- private final CuratorFramework client;
-
- /** Curator recipe for leader election. */
- private final LeaderLatch leaderLatch;
-
- /** Curator recipe to watch a given ZooKeeper node for changes. */
- private final TreeCache cache;
-
- /** ZooKeeper path of the node which stores the current leader information. */
- private final String connectionInformationPath;
-
- private final String leaderLatchPath;
-
- private final ConnectionStateListener listener =
- (client, newState) -> handleStateChange(newState);
-
- private final LeaderElectionEventHandler leaderElectionEventHandler;
-
- private final FatalErrorHandler fatalErrorHandler;
-
- private volatile boolean running;
-
- /**
- * Creates a ZooKeeperLeaderElectionDriver object.
- *
- * @param client Client which is connected to the ZooKeeper quorum
- * @param path ZooKeeper node path for the leader election
- * @param leaderElectionEventHandler Event handler for processing leader change events
- * @param fatalErrorHandler Fatal error handler
- */
- public ZooKeeperLeaderElectionDriver(
- CuratorFramework client,
- String path,
- LeaderElectionEventHandler leaderElectionEventHandler,
- FatalErrorHandler fatalErrorHandler)
- throws Exception {
- checkNotNull(path);
- this.client = checkNotNull(client);
- this.connectionInformationPath = ZooKeeperUtils.generateConnectionInformationPath(path);
- this.leaderElectionEventHandler = checkNotNull(leaderElectionEventHandler);
- this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
-
- leaderLatchPath = ZooKeeperUtils.generateLeaderLatchPath(path);
- leaderLatch = new LeaderLatch(client, leaderLatchPath);
- this.cache =
- ZooKeeperUtils.createTreeCache(
- client,
- connectionInformationPath,
- this::retrieveLeaderInformationFromZooKeeper);
-
- running = true;
-
- leaderLatch.addListener(this);
- leaderLatch.start();
-
- cache.start();
-
- client.getConnectionStateListenable().addListener(listener);
- }
-
- @Override
- public void close() throws Exception {
- if (!running) {
- return;
- }
- running = false;
-
- LOG.info("Closing {}", this);
-
- client.getConnectionStateListenable().removeListener(listener);
-
- Exception exception = null;
-
- try {
- cache.close();
- } catch (Exception e) {
- exception = e;
- }
-
- try {
- leaderLatch.close();
- } catch (Exception e) {
- exception = ExceptionUtils.firstOrSuppressed(e, exception);
- }
-
- if (exception != null) {
- throw new Exception(
- "Could not properly stop the ZooKeeperLeaderElectionDriver.", exception);
- }
- }
-
- @Override
- public boolean hasLeadership() {
- return running && leaderLatch.hasLeadership();
- }
-
- @Override
- public void isLeader() {
- leaderElectionEventHandler.onGrantLeadership(UUID.randomUUID());
- }
-
- @Override
- public void notLeader() {
- leaderElectionEventHandler.onRevokeLeadership();
- }
-
- private void retrieveLeaderInformationFromZooKeeper() throws Exception {
- if (leaderLatch.hasLeadership()) {
- ChildData childData = cache.getCurrentData(connectionInformationPath);
- leaderElectionEventHandler.onLeaderInformationChange(
- childData == null
- ? LeaderInformation.empty()
- : ZooKeeperUtils.readLeaderInformation(childData.getData()));
- }
- }
-
- /** Writes the current leader's address as well the given leader session ID to ZooKeeper. */
- @Override
- public void writeLeaderInformation(LeaderInformation leaderInformation) {
- Preconditions.checkState(
- running,
- "Leader information can only be written if the driver hasn't been closed, yet.");
-
- // this method does not have to be synchronized because the curator framework client
- // is thread-safe. We do not write the empty data to ZooKeeper here. Because
- // check-leadership-and-update
- // is not a transactional operation. We may wrongly clear the data written by new leader.
- if (LOG.isDebugEnabled()) {
- LOG.debug("Write leader information: {}.", leaderInformation);
- }
- if (leaderInformation.isEmpty()) {
- return;
- }
-
- try {
- ZooKeeperUtils.writeLeaderInformationToZooKeeper(
- leaderInformation,
- client,
- leaderLatch::hasLeadership,
- connectionInformationPath);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Successfully wrote leader information: {}.", leaderInformation);
- }
- } catch (Exception e) {
- fatalErrorHandler.onFatalError(
- new LeaderElectionException(
- "Could not write leader address and leader session ID to "
- + "ZooKeeper.",
- e));
- }
- }
-
- private void handleStateChange(ConnectionState newState) {
- switch (newState) {
- case CONNECTED:
- LOG.debug("Connected to ZooKeeper quorum. Leader election can start.");
- break;
- case SUSPENDED:
- LOG.warn("Connection to ZooKeeper suspended, waiting for reconnection.");
- break;
- case RECONNECTED:
- LOG.info(
- "Connection to ZooKeeper was reconnected. Leader election can be restarted.");
- break;
- case LOST:
- // Maybe we have to throw an exception here to terminate the JobManager
- LOG.warn(
- "Connection to ZooKeeper lost. The contender no longer participates in the leader election.");
- break;
- }
- }
-
- @Override
- public String toString() {
- return String.format(
- "%s{leaderLatchPath='%s', connectionInformationPath='%s'}",
- getClass().getSimpleName(), leaderLatchPath, connectionInformationPath);
- }
-
- @VisibleForTesting
- String getConnectionInformationPath() {
- return connectionInformationPath;
- }
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionDriverFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionDriverFactory.java
deleted file mode 100644
index eecbe25c01c..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionDriverFactory.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.leaderelection;
-
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
-
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework;
-
-/**
- * {@link LeaderElectionDriverFactory} implementation for Zookeeper.
- *
- * @deprecated in favour of {@link ZooKeeperMultipleComponentLeaderElectionDriverFactory}
- */
-public class ZooKeeperLeaderElectionDriverFactory implements LeaderElectionDriverFactory {
-
- private final CuratorFramework client;
-
- private final String path;
-
- public ZooKeeperLeaderElectionDriverFactory(CuratorFramework client, String path) {
- this.client = client;
- this.path = path;
- }
-
- @Override
- public ZooKeeperLeaderElectionDriver createLeaderElectionDriver(
- LeaderElectionEventHandler leaderEventHandler, FatalErrorHandler fatalErrorHandler)
- throws Exception {
- return new ZooKeeperLeaderElectionDriver(
- client, path, leaderEventHandler, fatalErrorHandler);
- }
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/LeaderRetrievalDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/LeaderRetrievalDriver.java
index 48add240540..338aa50794c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/LeaderRetrievalDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/LeaderRetrievalDriver.java
@@ -20,7 +20,8 @@ package org.apache.flink.runtime.leaderretrieval;
/**
* A {@link LeaderRetrievalDriver} is responsible for retrieves the current leader which has been
- * elected by the {@link org.apache.flink.runtime.leaderelection.LeaderElectionDriver}.
+ * elected by the {@link
+ * org.apache.flink.runtime.leaderelection.MultipleComponentLeaderElectionDriver}.
*
* <p><strong>Important</strong>: The {@link LeaderRetrievalDriver} could not guarantee that there
* is no {@link LeaderRetrievalEventHandler} callbacks happen after {@link #close()}.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalDriver.java
index ec1908e3185..762978fb694 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalDriver.java
@@ -20,7 +20,7 @@ package org.apache.flink.runtime.leaderretrieval;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.leaderelection.LeaderInformation;
-import org.apache.flink.runtime.leaderelection.ZooKeeperLeaderElectionDriver;
+import org.apache.flink.runtime.leaderelection.ZooKeeperMultipleComponentLeaderElectionDriver;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.util.ZooKeeperUtils;
import org.apache.flink.util.ExceptionUtils;
@@ -41,10 +41,10 @@ import java.util.UUID;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
- * The counterpart to the {@link ZooKeeperLeaderElectionDriver}. {@link LeaderRetrievalService}
- * implementation for Zookeeper. It retrieves the current leader which has been elected by the
- * {@link ZooKeeperLeaderElectionDriver}. The leader address as well as the current leader session
- * ID is retrieved from ZooKeeper.
+ * The counterpart to the {@link ZooKeeperMultipleComponentLeaderElectionDriver}. {@link
+ * LeaderRetrievalService} implementation for Zookeeper. It retrieves the current leader which has
+ * been elected by the {@link ZooKeeperMultipleComponentLeaderElectionDriver}. The leader address as
+ * well as the current leader session ID is retrieved from ZooKeeper.
*/
public class ZooKeeperLeaderRetrievalDriver implements LeaderRetrievalDriver {
private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperLeaderRetrievalDriver.class);
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 62e4504fb6d..da0402bbfa8 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
@@ -41,7 +41,6 @@ import java.util.function.BiConsumer;
import java.util.function.Function;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link DefaultLeaderElectionService}. */
@@ -130,8 +129,9 @@ class DefaultLeaderElectionServiceTest {
}
/**
- * Tests that we can shut down the DefaultLeaderElectionService if the used LeaderElectionDriver
- * holds an internal lock. See FLINK-20008 for more details.
+ * Tests that we can shut down the DefaultLeaderElectionService if the used {@link
+ * MultipleComponentLeaderElectionDriver} holds an internal lock. See FLINK-20008 for more
+ * details.
*/
@Test
void testCloseGrantDeadlock() throws Exception {
@@ -333,7 +333,7 @@ class DefaultLeaderElectionServiceTest {
try (final DefaultLeaderElectionService testInstance =
new DefaultLeaderElectionService(
driverFactory, fatalErrorHandlerExtension.getTestingFatalErrorHandler())) {
- driverBuilder.setCloseConsumer(lock -> testInstance.onRevokeLeadership());
+ driverBuilder.setCloseConsumer(lock -> testInstance.notLeader());
testInstance.startLeaderElectionBackend();
leadershipGranted.set(true);
@@ -748,50 +748,6 @@ class DefaultLeaderElectionServiceTest {
};
}
- /**
- * This code path is still used by {@link DefaultMultipleComponentLeaderElectionService} where
- * there's only a single contender registered with the {@link DefaultLeaderElectionService}.
- * That method should become unused with FLINK-32409.
- */
- @Test
- void testDeprecatedLeaderInformationChangedFailsForMoreThanOneContender() throws Exception {
- final AtomicReference<LeaderInformationRegister> storedLeaderInformation =
- new AtomicReference<>();
- new Context(storedLeaderInformation) {
- {
- runTestWithSynchronousEventHandling(
- () -> {
- grantLeadership();
-
- assertThatThrownBy(
- () ->
- leaderElectionService.onLeaderInformationChange(
- LeaderInformation.empty()))
- .isInstanceOf(UnsupportedOperationException.class);
-
- contenderContext1.leaderElection.close();
-
- storedLeaderInformation.set(LeaderInformationRegister.empty());
-
- assertThatNoException()
- .isThrownBy(
- () ->
- leaderElectionService.onLeaderInformationChange(
- LeaderInformation.empty()));
-
- assertThat(
- storedLeaderInformation
- .get()
- .hasLeaderInformation(
- contenderContext0.contenderID))
- .as(
- "The LeaderInformation for the registered contender should have been corrected.")
- .isTrue();
- });
- }
- };
- }
-
@Test
void testOnGrantLeadershipIsIgnoredAfterLeaderElectionBeingStop() throws Exception {
new Context() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultMultipleComponentLeaderElectionServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultMultipleComponentLeaderElectionServiceTest.java
deleted file mode 100644
index a55941deed7..00000000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultMultipleComponentLeaderElectionServiceTest.java
+++ /dev/null
@@ -1,448 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.leaderelection;
-
-import org.apache.flink.core.testutils.FlinkAssertions;
-import org.apache.flink.core.testutils.OneShotLatch;
-import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.concurrent.Executors;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
-import javax.annotation.Nullable;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for the {@link DefaultMultipleComponentLeaderElectionService}. */
-class DefaultMultipleComponentLeaderElectionServiceTest {
-
- @RegisterExtension
- final TestingFatalErrorHandlerExtension fatalErrorHandlerExtension =
- new TestingFatalErrorHandlerExtension();
-
- @Test
- void isLeaderInformsAllRegisteredLeaderElectionEventHandlers() throws Exception {
- final DefaultMultipleComponentLeaderElectionService leaderElectionService =
- createDefaultMultiplexingLeaderElectionService();
-
- try {
- final Collection<SimpleTestingLeaderElectionEventListener> eventListeners =
- Stream.generate(SimpleTestingLeaderElectionEventListener::new)
- .limit(4)
- .collect(Collectors.toList());
-
- int counter = 0;
- for (SimpleTestingLeaderElectionEventListener eventListener : eventListeners) {
- leaderElectionService.registerLeaderElectionEventHandler(
- String.valueOf(counter), eventListener);
- counter++;
- }
-
- leaderElectionService.isLeader(UUID.randomUUID());
-
- for (SimpleTestingLeaderElectionEventListener eventListener : eventListeners) {
- assertThat(eventListener.hasLeadership()).isTrue();
- }
- } finally {
- leaderElectionService.close();
- }
- }
-
- private DefaultMultipleComponentLeaderElectionService
- createDefaultMultiplexingLeaderElectionService() throws Exception {
- final TestingLeaderElectionDriver.Factory driverFactory =
- new TestingLeaderElectionDriver.Factory(
- TestingLeaderElectionDriver.newBuilder(new AtomicBoolean()));
- return createDefaultMultiplexingLeaderElectionService(driverFactory);
- }
-
- private DefaultMultipleComponentLeaderElectionService
- createDefaultMultiplexingLeaderElectionService(
- TestingLeaderElectionDriver.Factory driverFactory) throws Exception {
- return new DefaultMultipleComponentLeaderElectionService(
- fatalErrorHandlerExtension.getTestingFatalErrorHandler(),
- driverFactory,
- Executors.newDirectExecutorService());
- }
-
- @Test
- void notLeaderInformsAllRegisteredLeaderElectionEventHandlers() throws Exception {
- final DefaultMultipleComponentLeaderElectionService leaderElectionService =
- createDefaultMultiplexingLeaderElectionService();
-
- try {
- final Collection<SimpleTestingLeaderElectionEventListener> eventListeners =
- Stream.generate(SimpleTestingLeaderElectionEventListener::new)
- .limit(4)
- .collect(Collectors.toList());
-
- int counter = 0;
- for (SimpleTestingLeaderElectionEventListener eventListener : eventListeners) {
- leaderElectionService.registerLeaderElectionEventHandler(
- String.valueOf(counter), eventListener);
- counter++;
- }
-
- leaderElectionService.isLeader(UUID.randomUUID());
- leaderElectionService.notLeader();
-
- for (SimpleTestingLeaderElectionEventListener eventListener : eventListeners) {
- assertThat(eventListener.hasLeadership()).isFalse();
- }
- } finally {
- leaderElectionService.close();
- }
- }
-
- @Test
- void handleFatalError() throws Exception {
- final TestingLeaderElectionDriver.Factory driverFactory =
- new TestingLeaderElectionDriver.Factory(
- TestingLeaderElectionDriver.newBuilder(new AtomicBoolean()));
-
- final DefaultMultipleComponentLeaderElectionService leaderElectionService =
- createDefaultMultiplexingLeaderElectionService(driverFactory);
- final TestingLeaderElectionDriver leaderElectionDriver =
- driverFactory.assertAndGetOnlyCreatedDriver();
-
- try {
- final Throwable expectedFatalError =
- new Exception("Expected exception simulating a fatal error.");
-
- leaderElectionDriver.triggerFatalError(expectedFatalError);
-
- FlinkAssertions.assertThatFuture(
- fatalErrorHandlerExtension
- .getTestingFatalErrorHandler()
- .getErrorFuture())
- .eventuallySucceeds()
- .isEqualTo(expectedFatalError);
- } finally {
- leaderElectionService.close();
- fatalErrorHandlerExtension.getTestingFatalErrorHandler().clearError();
- }
- }
-
- @Test
- void unregisteredEventHandlersAreNotNotified() throws Exception {
- final TestingLeaderElectionDriver.Factory driverFactory =
- new TestingLeaderElectionDriver.Factory(
- TestingLeaderElectionDriver.newBuilder(new AtomicBoolean()));
-
- final DefaultMultipleComponentLeaderElectionService leaderElectionService =
- createDefaultMultiplexingLeaderElectionService(driverFactory);
-
- try {
- final SimpleTestingLeaderElectionEventListener leaderElectionEventHandler =
- new SimpleTestingLeaderElectionEventListener();
- final String componentId = "foobar";
- leaderElectionService.registerLeaderElectionEventHandler(
- componentId, leaderElectionEventHandler);
- leaderElectionService.unregisterLeaderElectionEventHandler(componentId);
-
- leaderElectionService.isLeader(UUID.randomUUID());
-
- assertThat(leaderElectionEventHandler.hasLeadership()).isFalse();
- } finally {
- leaderElectionService.close();
- }
- }
-
- @Test
- void newlyRegisteredEventHandlersAreInformedAboutLeadership() throws Exception {
- final TestingLeaderElectionDriver.Factory driverFactory =
- new TestingLeaderElectionDriver.Factory(
- TestingLeaderElectionDriver.newBuilder(new AtomicBoolean()));
-
- final DefaultMultipleComponentLeaderElectionService leaderElectionService =
- createDefaultMultiplexingLeaderElectionService(driverFactory);
-
- try {
- leaderElectionService.isLeader(UUID.randomUUID());
-
- final SimpleTestingLeaderElectionEventListener leaderElectionEventHandler =
- new SimpleTestingLeaderElectionEventListener();
- leaderElectionService.registerLeaderElectionEventHandler(
- "foobar", leaderElectionEventHandler);
-
- assertThat(leaderElectionEventHandler.hasLeadership()).isTrue();
- } finally {
- leaderElectionService.close();
- }
- }
-
- @Test
- public void testLeaderSessionIdMatchesBetweenComponents() throws Exception {
- final TestingLeaderElectionDriver.Factory driverFactory =
- new TestingLeaderElectionDriver.Factory(
- TestingLeaderElectionDriver.newBuilder(new AtomicBoolean()));
-
- final DefaultMultipleComponentLeaderElectionService leaderElectionService =
- createDefaultMultiplexingLeaderElectionService(driverFactory);
-
- try {
- final Component preLeadershipGrantedComponent =
- new Component(
- "test-component-0",
- new SimpleTestingLeaderElectionEventListener(),
- LeaderInformation.empty());
- final Component postLeadershipGrantedComponent =
- new Component(
- "test-component-1",
- new SimpleTestingLeaderElectionEventListener(),
- LeaderInformation.empty());
-
- leaderElectionService.registerLeaderElectionEventHandler(
- preLeadershipGrantedComponent.getComponentId(),
- preLeadershipGrantedComponent.getLeaderElectionEventListener());
-
- leaderElectionService.isLeader(UUID.randomUUID());
-
- leaderElectionService.registerLeaderElectionEventHandler(
- postLeadershipGrantedComponent.getComponentId(),
- postLeadershipGrantedComponent.getLeaderElectionEventListener());
-
- final UUID preGrantLeaderInformation =
- preLeadershipGrantedComponent
- .getLeaderElectionEventListener()
- .getLeaderSessionID();
-
- final UUID postGrantLeaderInformation =
- postLeadershipGrantedComponent
- .getLeaderElectionEventListener()
- .getLeaderSessionID();
-
- assertThat(preGrantLeaderInformation).isEqualTo(postGrantLeaderInformation);
- } finally {
- leaderElectionService.close();
- }
- }
-
- @Test
- void allKnownLeaderInformationCallsEventHandlers() throws Exception {
- final TestingLeaderElectionDriver.Factory driverFactory =
- new TestingLeaderElectionDriver.Factory(
- TestingLeaderElectionDriver.newBuilder(new AtomicBoolean()));
-
- final DefaultMultipleComponentLeaderElectionService leaderElectionService =
- createDefaultMultiplexingLeaderElectionService(driverFactory);
-
- try {
- leaderElectionService.isLeader(UUID.randomUUID());
-
- final Collection<Component> knownLeaderInformation = createComponents(3);
- final Collection<Component> unknownLeaderInformation = createComponents(2);
-
- registerLeaderElectionEventHandler(leaderElectionService, knownLeaderInformation);
- registerLeaderElectionEventHandler(leaderElectionService, unknownLeaderInformation);
-
- leaderElectionService.notifyAllKnownLeaderInformation(
- new LeaderInformationRegister(
- knownLeaderInformation.stream()
- .collect(
- Collectors.toMap(
- Component::getComponentId,
- Component::getLeaderInformation))));
-
- for (Component component : knownLeaderInformation) {
- assertThat(component.getLeaderElectionEventListener().getLeaderInformation())
- .isEqualTo(component.getLeaderInformation());
- }
-
- for (Component component : unknownLeaderInformation) {
- assertThat(component.getLeaderElectionEventListener().getLeaderInformation())
- .isEqualTo(LeaderInformation.empty());
- }
-
- } finally {
- leaderElectionService.close();
- }
- }
-
- @Test
- void allKnownLeaderInformationDoesNotBlock() throws Exception {
- final TestingLeaderElectionDriver.Factory driverFactory =
- new TestingLeaderElectionDriver.Factory(
- TestingLeaderElectionDriver.newBuilder(new AtomicBoolean()));
-
- final DefaultMultipleComponentLeaderElectionService leaderElectionService =
- new DefaultMultipleComponentLeaderElectionService(
- fatalErrorHandlerExtension.getTestingFatalErrorHandler(),
- driverFactory,
- java.util.concurrent.Executors.newSingleThreadScheduledExecutor());
-
- try {
- leaderElectionService.isLeader(UUID.randomUUID());
-
- final String knownLeaderInformationComponent = "knownLeaderInformationComponent";
- final BlockingLeaderElectionEventHandler knownLeaderElectionEventHandler =
- new BlockingLeaderElectionEventHandler();
- leaderElectionService.registerLeaderElectionEventHandler(
- knownLeaderInformationComponent, knownLeaderElectionEventHandler);
- final BlockingLeaderElectionEventHandler unknownLeaderElectionEventHandler =
- new BlockingLeaderElectionEventHandler();
- leaderElectionService.registerLeaderElectionEventHandler(
- "unknownLeaderInformationComponent", unknownLeaderElectionEventHandler);
-
- // make sure that this call succeeds w/o blocking
- leaderElectionService.notifyAllKnownLeaderInformation(
- LeaderInformationRegister.of(
- knownLeaderInformationComponent,
- LeaderInformation.known(UUID.randomUUID(), "localhost")));
-
- knownLeaderElectionEventHandler.unblock();
- unknownLeaderElectionEventHandler.unblock();
- } finally {
- leaderElectionService.close();
- }
- }
-
- private static final class BlockingLeaderElectionEventHandler
- implements LeaderElectionEventHandler {
-
- private final OneShotLatch waitingLatch = new OneShotLatch();
-
- @Override
- public void onGrantLeadership(UUID newLeaderSessionId) {}
-
- @Override
- public void onRevokeLeadership() {}
-
- @Override
- public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- try {
- waitingLatch.await();
- } catch (InterruptedException e) {
- ExceptionUtils.checkInterrupted(e);
- }
- }
-
- void unblock() {
- waitingLatch.trigger();
- }
- }
-
- private void registerLeaderElectionEventHandler(
- DefaultMultipleComponentLeaderElectionService leaderElectionService,
- Collection<Component> knownLeaderInformation) {
- for (Component component : knownLeaderInformation) {
- leaderElectionService.registerLeaderElectionEventHandler(
- component.getComponentId(), component.getLeaderElectionEventListener());
- }
- }
-
- private Collection<Component> createComponents(int numberComponents) {
- final List<Component> result = new ArrayList<>();
-
- for (int i = 0; i < numberComponents; i++) {
- result.add(
- new Component(
- UUID.randomUUID().toString(),
- new SimpleTestingLeaderElectionEventListener(),
- LeaderInformation.known(UUID.randomUUID(), "localhost")));
- }
-
- return result;
- }
-
- private static final class Component {
- private final String componentId;
- private final SimpleTestingLeaderElectionEventListener leaderElectionEventListener;
- private final LeaderInformation leaderInformation;
-
- private Component(
- String componentId,
- SimpleTestingLeaderElectionEventListener leaderElectionEventListener,
- LeaderInformation leaderInformation) {
- this.componentId = componentId;
- this.leaderElectionEventListener = leaderElectionEventListener;
- this.leaderInformation = leaderInformation;
- }
-
- String getComponentId() {
- return componentId;
- }
-
- LeaderInformation getLeaderInformation() {
- return leaderInformation;
- }
-
- SimpleTestingLeaderElectionEventListener getLeaderElectionEventListener() {
- return leaderElectionEventListener;
- }
- }
-
- private static final class SimpleTestingLeaderElectionEventListener
- implements LeaderElectionEventHandler {
-
- /**
- * {@code currentLeaderSessionId} is set if the current LeaderElection client is select as
- * the leader by the HA backend.
- */
- @Nullable private UUID currentLeaderSessionId;
-
- /** {@code leaderInformation} is set if the leader information in the HA backend changes. */
- @Nullable private LeaderInformation leaderInformation;
-
- SimpleTestingLeaderElectionEventListener() {
- currentLeaderSessionId = null;
- leaderInformation = null;
- }
-
- public boolean hasLeadership() {
- return currentLeaderSessionId != null;
- }
-
- @Override
- public void onGrantLeadership(UUID newLeaderSessionId) {
- currentLeaderSessionId = newLeaderSessionId;
- }
-
- @Override
- public void onRevokeLeadership() {
- currentLeaderSessionId = null;
- leaderInformation = null;
- }
-
- @Override
- public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- this.leaderInformation = leaderInformation;
- }
-
- @Nullable
- LeaderInformation getLeaderInformation() {
- return leaderInformation;
- }
-
- @Nullable
- UUID getLeaderSessionID() {
- return currentLeaderSessionId;
- }
- }
-}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderBase.java
index c504e3eda2e..9b21b743305 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderBase.java
@@ -28,7 +28,7 @@ import java.util.concurrent.LinkedBlockingQueue;
/**
* Base class which provides some convenience functions for testing purposes of {@link
- * LeaderContender} and {@link LeaderElectionEventHandler}.
+ * LeaderContender}.
*/
public class TestingLeaderBase {
// The queues will be offered by subclasses
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 cb69d177eb6..ed32afefd89 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
@@ -380,8 +380,8 @@ class ZooKeeperLeaderElectionTest {
}
/**
- * Test that errors in the {@link LeaderElectionDriver} are correctly forwarded to the {@link
- * LeaderContender}.
+ * Test that errors in the {@link MultipleComponentLeaderElectionDriver} are correctly forwarded
+ * to the {@link LeaderContender}.
*/
@Test
void testExceptionForwarding() throws Exception {
@@ -552,8 +552,8 @@ class ZooKeeperLeaderElectionTest {
}
/**
- * Test that background errors in the {@link LeaderElectionDriver} are correctly forwarded to
- * the {@link FatalErrorHandler}.
+ * Test that background errors in the {@link MultipleComponentLeaderElectionDriver} are
+ * correctly forwarded to the {@link FatalErrorHandler}.
*/
@Test
public void testUnExpectedErrorForwarding() throws Exception {