This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ba027b6b1a9 [FLINK-25806][ha] Remove legacy HA services
ba027b6b1a9 is described below
commit ba027b6b1a956b425ff14ed8b55c6aeef3e565c8
Author: Chesnay Schepler <[email protected]>
AuthorDate: Tue Apr 12 13:20:42 2022 +0200
[FLINK-25806][ha] Remove legacy HA services
---
.../expert_high_availability_section.html | 6 -
.../generated/high_availability_configuration.html | 6 -
.../configuration/HighAvailabilityOptions.java | 19 --
.../highavailability/KubernetesHaServices.java | 197 -------------
.../KubernetesHaServicesFactory.java | 31 +--
.../kubernetes/kubeclient/FlinkKubeClient.java | 4 +-
.../highavailability/KubernetesHaServicesTest.java | 113 --------
.../HighAvailabilityServicesUtils.java | 21 +-
.../zookeeper/ZooKeeperHaServices.java | 133 ---------
.../zookeeper/ZooKeeperHaServicesTest.java | 308 ---------------------
.../ZooKeeperLeaderRetrievalTest.java | 9 +-
11 files changed, 22 insertions(+), 825 deletions(-)
diff --git
a/docs/layouts/shortcodes/generated/expert_high_availability_section.html
b/docs/layouts/shortcodes/generated/expert_high_availability_section.html
index c6fb138cd9a..0f0391fb3cc 100644
--- a/docs/layouts/shortcodes/generated/expert_high_availability_section.html
+++ b/docs/layouts/shortcodes/generated/expert_high_availability_section.html
@@ -14,11 +14,5 @@
<td>String</td>
<td>The port (range) used by the Flink Master for its RPC
connections in highly-available setups. In highly-available setups, this value
is used instead of 'jobmanager.rpc.port'.A value of '0' means that a random
free port is chosen. TaskManagers discover this port through the
high-availability services (leader election), so a random port or a port range
works without requiring any additional means of service discovery.</td>
</tr>
- <tr>
- <td><h5>high-availability.use-old-ha-services</h5></td>
- <td style="word-wrap: break-word;">false</td>
- <td>Boolean</td>
- <td>Use this option to disable the new HA service implementations
for ZooKeeper and K8s. This is a safety hatch in case that the new ha services
are buggy.</td>
- </tr>
</tbody>
</table>
diff --git
a/docs/layouts/shortcodes/generated/high_availability_configuration.html
b/docs/layouts/shortcodes/generated/high_availability_configuration.html
index 7c3ec2243ec..324f3378fb1 100644
--- a/docs/layouts/shortcodes/generated/high_availability_configuration.html
+++ b/docs/layouts/shortcodes/generated/high_availability_configuration.html
@@ -32,12 +32,6 @@
<td>String</td>
<td>File system path (URI) where Flink persists metadata in
high-availability setups.</td>
</tr>
- <tr>
- <td><h5>high-availability.use-old-ha-services</h5></td>
- <td style="word-wrap: break-word;">false</td>
- <td>Boolean</td>
- <td>Use this option to disable the new HA service implementations
for ZooKeeper and K8s. This is a safety hatch in case that the new ha services
are buggy.</td>
- </tr>
<tr>
<td><h5>high-availability.zookeeper.client.acl</h5></td>
<td style="word-wrap: break-word;">"open"</td>
diff --git
a/flink-core/src/main/java/org/apache/flink/configuration/HighAvailabilityOptions.java
b/flink-core/src/main/java/org/apache/flink/configuration/HighAvailabilityOptions.java
index a7a173facbf..b156c152d9e 100644
---
a/flink-core/src/main/java/org/apache/flink/configuration/HighAvailabilityOptions.java
+++
b/flink-core/src/main/java/org/apache/flink/configuration/HighAvailabilityOptions.java
@@ -217,25 +217,6 @@ public class HighAvailabilityOptions {
.withDescription(
"The time before a JobManager after a fail over
recovers the current jobs.");
- /**
- * Safety hatch to fallback to the old ha services implementations.
- *
- * <p>Ideally, we can remove this option together with the old
implementations in the next
- * release.
- *
- * @see <a
href="https://issues.apache.org/jira/browse/FLINK-25806";>FLINK-25806</a>
- */
- @Documentation.Section(Documentation.Sections.EXPERT_HIGH_AVAILABILITY)
- public static final ConfigOption<Boolean> USE_OLD_HA_SERVICES =
- key("high-availability.use-old-ha-services")
- .booleanType()
- .defaultValue(false)
- .withDescription(
- Description.builder()
- .text(
- "Use this option to disable the
new HA service implementations for ZooKeeper and K8s. This is a safety hatch in
case that the new ha services are buggy.")
- .build());
-
// ------------------------------------------------------------------------
/** Not intended to be instantiated. */
diff --git
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServices.java
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServices.java
deleted file mode 100644
index 1a6b6aadfe6..00000000000
---
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServices.java
+++ /dev/null
@@ -1,197 +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.api.common.JobID;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
-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.utils.KubernetesUtils;
-import org.apache.flink.runtime.blob.BlobStoreService;
-import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
-import org.apache.flink.runtime.highavailability.AbstractHaServices;
-import org.apache.flink.runtime.highavailability.FileSystemJobResultStore;
-import org.apache.flink.runtime.jobmanager.JobGraphStore;
-import org.apache.flink.runtime.leaderelection.DefaultLeaderElectionService;
-import org.apache.flink.runtime.leaderelection.LeaderElectionService;
-import org.apache.flink.runtime.leaderretrieval.DefaultLeaderRetrievalService;
-import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
-import org.apache.flink.util.ExecutorUtils;
-import org.apache.flink.util.concurrent.ExecutorThreadFactory;
-
-import java.io.IOException;
-import java.util.UUID;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-
-import static
org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
-import static org.apache.flink.kubernetes.utils.Constants.NAME_SEPARATOR;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * An implementation of the {@link AbstractHaServices} using Kubernetes.
- *
- * <p>All the HA information relevant for a specific component will be stored
in a single ConfigMap.
- * For example, the Dispatcher's ConfigMap would then contain the current
leader, the running jobs
- * and the pointers to the persisted JobGraphs. The JobManager's ConfigMap
would then contain the
- * current leader, the pointers to the checkpoints and the checkpoint ID
counter.
- *
- * <p>The ConfigMap name will be created with the pattern
"{clusterId}-{componentName}-leader".
- * Given that the cluster id is configured to "k8s-ha-app1", then we could get
the following
- * ConfigMap names. e.g. k8s-ha-app1-restserver-leader,
- * k8s-ha-app1-00000000000000000000000000000000-jobmanager-leader
- *
- * <p>Note that underline("_") is not allowed in Kubernetes ConfigMap name.
- *
- * @deprecated in favour of {@link
KubernetesMultipleComponentLeaderElectionHaServices}
- */
-@Deprecated
-public class KubernetesHaServices extends AbstractHaServices {
-
- private final String clusterId;
-
- /** Kubernetes client. */
- private final FlinkKubeClient kubeClient;
-
- private final KubernetesConfigMapSharedWatcher configMapSharedWatcher;
- private final ExecutorService watchExecutorService;
-
- private static final String RESOURCE_MANAGER_NAME = "resourcemanager";
-
- private static final String DISPATCHER_NAME = "dispatcher";
-
- private static final String JOB_MANAGER_NAME = "jobmanager";
-
- private static final String REST_SERVER_NAME = "restserver";
-
- private static final String LEADER_SUFFIX = "leader";
-
- /**
- * Each {@link KubernetesHaServices} will have a dedicated lock identity
for all the components
- * above. Different instances will have different identities.
- */
- private final String lockIdentity;
-
- KubernetesHaServices(
- FlinkKubeClient kubeClient,
- Executor executor,
- Configuration config,
- BlobStoreService blobStoreService)
- throws IOException {
- super(
- config,
- executor,
- blobStoreService,
- FileSystemJobResultStore.fromConfiguration(config));
-
- this.kubeClient = checkNotNull(kubeClient);
- this.clusterId =
checkNotNull(config.get(KubernetesConfigOptions.CLUSTER_ID));
-
- this.configMapSharedWatcher =
- this.kubeClient.createConfigMapSharedWatcher(
- KubernetesUtils.getConfigMapLabels(
- clusterId,
LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY));
- this.watchExecutorService =
- Executors.newCachedThreadPool(
- new ExecutorThreadFactory("config-map-watch-handler"));
-
- lockIdentity = UUID.randomUUID().toString();
- }
-
- @Override
- public LeaderElectionService createLeaderElectionService(String
leaderName) {
- final KubernetesLeaderElectionConfiguration leaderConfig =
- new KubernetesLeaderElectionConfiguration(leaderName,
lockIdentity, configuration);
- return new DefaultLeaderElectionService(
- new KubernetesLeaderElectionDriverFactory(
- kubeClient, configMapSharedWatcher,
watchExecutorService, leaderConfig));
- }
-
- @Override
- public LeaderRetrievalService createLeaderRetrievalService(String
leaderName) {
- return new DefaultLeaderRetrievalService(
- new KubernetesLeaderRetrievalDriverFactory(
- kubeClient, configMapSharedWatcher,
watchExecutorService, leaderName));
- }
-
- @Override
- public CheckpointRecoveryFactory createCheckpointRecoveryFactory() {
- return KubernetesCheckpointRecoveryFactory.withLeadershipValidation(
- kubeClient,
- configuration,
- ioExecutor,
- clusterId,
- this::getLeaderPathForJobManager,
- lockIdentity);
- }
-
- @Override
- public JobGraphStore createJobGraphStore() throws Exception {
- return KubernetesUtils.createJobGraphStore(
- configuration, kubeClient, getLeaderPathForDispatcher(),
lockIdentity);
- }
-
- @Override
- public void internalClose() {
- configMapSharedWatcher.close();
- kubeClient.close();
- ExecutorUtils.gracefulShutdown(5, TimeUnit.SECONDS,
this.watchExecutorService);
- }
-
- @Override
- public void internalCleanup() throws Exception {
- kubeClient
- .deleteConfigMapsByLabels(
- KubernetesUtils.getConfigMapLabels(
- clusterId,
LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY))
- .get();
- }
-
- @Override
- public void internalCleanupJobData(JobID jobID) throws Exception {
- kubeClient.deleteConfigMap(getLeaderPathForJobManager(jobID)).get();
- }
-
- @Override
- protected String getLeaderPathForResourceManager() {
- return getLeaderName(RESOURCE_MANAGER_NAME);
- }
-
- @Override
- protected String getLeaderPathForDispatcher() {
- return getLeaderName(DISPATCHER_NAME);
- }
-
- public String getLeaderPathForJobManager(final JobID jobID) {
- return getLeaderName(jobID.toString() + NAME_SEPARATOR +
JOB_MANAGER_NAME);
- }
-
- @Override
- protected String getLeaderPathForRestServer() {
- return getLeaderName(REST_SERVER_NAME);
- }
-
- private String getLeaderName(String component) {
- return clusterId + NAME_SEPARATOR + component + NAME_SEPARATOR +
LEADER_SUFFIX;
- }
-}
diff --git
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServicesFactory.java
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServicesFactory.java
index cee4402256c..00e61c3e0d3 100644
---
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServicesFactory.java
+++
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServicesFactory.java
@@ -19,7 +19,6 @@
package org.apache.flink.kubernetes.highavailability;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.kubernetes.kubeclient.FlinkKubeClientFactory;
import org.apache.flink.runtime.blob.BlobUtils;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
@@ -34,26 +33,14 @@ public class KubernetesHaServicesFactory implements
HighAvailabilityServicesFact
@Override
public HighAvailabilityServices createHAServices(Configuration
configuration, Executor executor)
throws Exception {
- final boolean useOldHaServices =
- configuration.get(HighAvailabilityOptions.USE_OLD_HA_SERVICES);
-
- if (useOldHaServices) {
- return new KubernetesHaServices(
- FlinkKubeClientFactory.getInstance()
- .fromConfiguration(configuration,
"kubernetes-ha-services"),
- executor,
- configuration,
- BlobUtils.createBlobStoreFromConfig(configuration));
- } else {
- return new KubernetesMultipleComponentLeaderElectionHaServices(
- FlinkKubeClientFactory.getInstance()
- .fromConfiguration(configuration,
"kubernetes-ha-services"),
- executor,
- configuration,
- BlobUtils.createBlobStoreFromConfig(configuration),
- error ->
-
FatalExitExceptionHandler.INSTANCE.uncaughtException(
- Thread.currentThread(), error));
- }
+ return new KubernetesMultipleComponentLeaderElectionHaServices(
+ FlinkKubeClientFactory.getInstance()
+ .fromConfiguration(configuration,
"kubernetes-ha-services"),
+ executor,
+ configuration,
+ BlobUtils.createBlobStoreFromConfig(configuration),
+ error ->
+ FatalExitExceptionHandler.INSTANCE.uncaughtException(
+ Thread.currentThread(), error));
}
}
diff --git
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
index d44f7159bbb..47038244a3f 100644
---
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
+++
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
@@ -167,8 +167,8 @@ public interface FlinkKubeClient extends AutoCloseable {
Function<KubernetesConfigMap, Optional<KubernetesConfigMap>>
updateFunction);
/**
- * Delete the Kubernetes ConfigMaps by labels. This will be used by {@link
- * org.apache.flink.kubernetes.highavailability.KubernetesHaServices} to
clean up all data.
+ * Delete the Kubernetes ConfigMaps by labels. This will be used by the HA
service to clean up
+ * all data.
*
* @param labels labels to filter the resources. e.g. type:
high-availability
* @return Return the delete future that only completes successfully, if
the resources that are
diff --git
a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServicesTest.java
b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServicesTest.java
deleted file mode 100644
index f2233ababcd..00000000000
---
a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServicesTest.java
+++ /dev/null
@@ -1,113 +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.api.common.JobID;
-import org.apache.flink.kubernetes.kubeclient.TestingFlinkKubeClient;
-import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
-import org.apache.flink.runtime.blob.VoidBlobStore;
-
-import org.junit.Test;
-
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-import static
org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
-import static
org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_KEY;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
-
-/** Tests for the {@link KubernetesHaServices}. */
-public class KubernetesHaServicesTest extends
KubernetesHighAvailabilityTestBase {
-
- @Test
- public void testInternalCloseShouldCloseFlinkKubeClient() throws Exception
{
- new Context() {
- {
- runTest(
- () -> {
- final KubernetesHaServices kubernetesHaServices =
- new KubernetesHaServices(
- flinkKubeClient,
- executorService,
- configuration,
- new VoidBlobStore());
- kubernetesHaServices.internalClose();
- assertThat(closeKubeClientFuture.isDone(),
is(true));
- });
- }
- };
- }
-
- @Test
- public void testInternalCleanupShouldCleanupConfigMaps() throws Exception {
- new Context() {
- {
- runTest(
- () -> {
- final KubernetesHaServices kubernetesHaServices =
- new KubernetesHaServices(
- flinkKubeClient,
- executorService,
- configuration,
- new VoidBlobStore());
- kubernetesHaServices.internalCleanup();
- final Map<String, String> labels =
- deleteConfigMapByLabelsFuture.get(
- TIMEOUT, TimeUnit.MILLISECONDS);
- assertThat(labels.size(), is(3));
- assertThat(
- labels.get(LABEL_CONFIGMAP_TYPE_KEY),
-
is(LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY));
- });
- }
- };
- }
-
- @Test
- public void testInternalJobCleanupShouldCleanupConfigMaps() throws
Exception {
- new Context() {
- {
- runTest(
- () -> {
- final KubernetesHaServices kubernetesHaServices =
- new KubernetesHaServices(
- flinkKubeClient,
- executorService,
- configuration,
- new VoidBlobStore());
- JobID jobID = new JobID();
- String configMapName =
-
kubernetesHaServices.getLeaderPathForJobManager(jobID);
- final KubernetesConfigMap configMap =
- new
TestingFlinkKubeClient.MockKubernetesConfigMap(
- configMapName);
- flinkKubeClient.createConfigMap(configMap);
- assertThat(
-
flinkKubeClient.getConfigMap(configMapName).isPresent(),
- is(true));
- kubernetesHaServices.internalCleanupJobData(jobID);
- assertThat(
-
flinkKubeClient.getConfigMap(configMapName).isPresent(),
- is(false));
- });
- }
- };
- }
-}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
index 44822648335..b54b86dda23 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
@@ -34,7 +34,6 @@ import
org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClie
import
org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneHaServices;
import
org.apache.flink.runtime.highavailability.zookeeper.CuratorFrameworkWithUnhandledErrorListener;
import
org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperClientHAServices;
-import org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServices;
import
org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperMultipleComponentLeaderElectionHaServices;
import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
import org.apache.flink.runtime.resourcemanager.ResourceManager;
@@ -81,25 +80,17 @@ public class HighAvailabilityServicesUtils {
private static HighAvailabilityServices createZooKeeperHaServices(
Configuration configuration, Executor executor, FatalErrorHandler
fatalErrorHandler)
throws Exception {
- final boolean useOldHaServices =
- configuration.get(HighAvailabilityOptions.USE_OLD_HA_SERVICES);
-
BlobStoreService blobStoreService =
BlobUtils.createBlobStoreFromConfig(configuration);
final CuratorFrameworkWithUnhandledErrorListener
curatorFrameworkWrapper =
ZooKeeperUtils.startCuratorFramework(configuration,
fatalErrorHandler);
- if (useOldHaServices) {
- return new ZooKeeperHaServices(
- curatorFrameworkWrapper, executor, configuration,
blobStoreService);
- } else {
- return new ZooKeeperMultipleComponentLeaderElectionHaServices(
- curatorFrameworkWrapper,
- configuration,
- executor,
- blobStoreService,
- fatalErrorHandler);
- }
+ return new ZooKeeperMultipleComponentLeaderElectionHaServices(
+ curatorFrameworkWrapper,
+ configuration,
+ executor,
+ blobStoreService,
+ fatalErrorHandler);
}
public static HighAvailabilityServices createHighAvailabilityServices(
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServices.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServices.java
deleted file mode 100644
index b1b05106845..00000000000
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServices.java
+++ /dev/null
@@ -1,133 +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.highavailability.zookeeper;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.HighAvailabilityOptions;
-import org.apache.flink.runtime.blob.BlobStoreService;
-import org.apache.flink.runtime.highavailability.AbstractHaServices;
-import org.apache.flink.runtime.leaderelection.LeaderElectionService;
-import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
-import org.apache.flink.runtime.util.ZooKeeperUtils;
-
-import java.io.IOException;
-import java.util.concurrent.Executor;
-
-/**
- * An implementation of the {@link AbstractHaServices} using Apache ZooKeeper.
The services store
- * data in ZooKeeper's nodes as illustrated by the following tree structure:
- *
- * <pre>
- * /flink
- * +/cluster_id_1/leader/resource_manager/latch
- * | | /connection_info
- * | | /dispatcher/latch
- * | | /connection_info
- * | | /rest_server/latch
- * | | /connection_info
- * | |
- * | |
- * | +jobgraphs/job-id-1
- * | | /job-id-2
- * | +jobs/job-id-1/leader/latch
- * | | /connection_info
- * | | /checkpoints/latest
- * | | /latest-1
- * | | /latest-2
- * | | /checkpoint_id_counter
- * |
- * +/cluster_id_2/leader/resource_manager/latch
- * | | /connection_info
- * | | /dispatcher/latch
- * | | /connection_info
- * | | /rest_server/latch
- * | | /connection_info
- * | |
- * | +jobgraphs/job-id-2
- * | +jobs/job-id-2/leader/latch
- * | | /connection_info
- * | | /checkpoints/latest
- * | | /latest-1
- * | | /latest-2
- * | | /checkpoint_id_counter
- * </pre>
- *
- * <p>The root path "/flink" is configurable via the option {@link
- * HighAvailabilityOptions#HA_ZOOKEEPER_ROOT}. This makes sure Flink stores
its data under specific
- * subtrees in ZooKeeper, for example to accommodate specific permission.
- *
- * <p>The "cluster_id" part identifies the data stored for a specific Flink
"cluster". This
- * "cluster" can be either a standalone or containerized Flink cluster, or it
can be job on a
- * framework like YARN (in a "per-job-cluster" mode).
- *
- * <p>In case of a "per-job-cluster" on YARN, the cluster-id is generated and
configured
- * automatically by the client or dispatcher that submits the Job to YARN.
- *
- * <p>In the case of a standalone cluster, that cluster-id needs to be
configured via {@link
- * HighAvailabilityOptions#HA_CLUSTER_ID}. All nodes with the same cluster id
will join the same
- * cluster and participate in the execution of the same set of jobs.
- *
- * @deprecated in favour of {@link
ZooKeeperMultipleComponentLeaderElectionHaServices}
- */
-@Deprecated
-public class ZooKeeperHaServices extends AbstractZooKeeperHaServices {
-
- // ------------------------------------------------------------------------
-
- public ZooKeeperHaServices(
- CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper,
- Executor executor,
- Configuration configuration,
- BlobStoreService blobStoreService)
- throws IOException {
- super(curatorFrameworkWrapper, executor, configuration,
blobStoreService);
- }
-
- @Override
- protected LeaderElectionService createLeaderElectionService(String
leaderPath) {
- return
ZooKeeperUtils.createLeaderElectionService(getCuratorFramework(), leaderPath);
- }
-
- @Override
- protected LeaderRetrievalService createLeaderRetrievalService(String
leaderPath) {
- return ZooKeeperUtils.createLeaderRetrievalService(
- getCuratorFramework(), leaderPath, configuration);
- }
-
- @Override
- protected String getLeaderPathForResourceManager() {
- return ZooKeeperUtils.getLeaderPathForResourceManager();
- }
-
- @Override
- protected String getLeaderPathForDispatcher() {
- return ZooKeeperUtils.getLeaderPathForDispatcher();
- }
-
- @Override
- public String getLeaderPathForJobManager(final JobID jobID) {
- return ZooKeeperUtils.getLeaderPathForJobManager(jobID);
- }
-
- @Override
- protected String getLeaderPathForRestServer() {
- return ZooKeeperUtils.getLeaderPathForRestServer();
- }
-}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServicesTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServicesTest.java
deleted file mode 100644
index 54165f39c83..00000000000
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServicesTest.java
+++ /dev/null
@@ -1,308 +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.highavailability.zookeeper;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.HighAvailabilityOptions;
-import org.apache.flink.runtime.blob.BlobKey;
-import org.apache.flink.runtime.blob.BlobStoreService;
-import org.apache.flink.runtime.leaderelection.LeaderElectionService;
-import org.apache.flink.runtime.leaderelection.TestingContender;
-import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
-import org.apache.flink.runtime.util.LeaderRetrievalUtils;
-import org.apache.flink.runtime.util.TestingFatalErrorHandlerResource;
-import org.apache.flink.runtime.util.ZooKeeperUtils;
-import org.apache.flink.runtime.zookeeper.ZooKeeperResource;
-import org.apache.flink.util.TestLogger;
-import org.apache.flink.util.concurrent.Executors;
-import org.apache.flink.util.function.ThrowingConsumer;
-
-import
org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework;
-import
org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.flink.shaded.curator5.org.apache.curator.retry.RetryNTimes;
-
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
-
-import javax.annotation.Nonnull;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.List;
-
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.hasItem;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.not;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.hamcrest.Matchers.nullValue;
-import static org.junit.Assert.assertThat;
-
-/** Tests for the {@link ZooKeeperHaServices}. */
-public class ZooKeeperHaServicesTest extends TestLogger {
-
- @ClassRule public static final ZooKeeperResource ZOO_KEEPER_RESOURCE = new
ZooKeeperResource();
-
- @Rule
- public final TestingFatalErrorHandlerResource
testingFatalErrorHandlerResource =
- new TestingFatalErrorHandlerResource();
-
- private static CuratorFramework client;
-
- @BeforeClass
- public static void setupClass() {
- client = startCuratorFramework();
- client.start();
- }
-
- @Before
- public void setup() throws Exception {
- final List<String> children = client.getChildren().forPath("/");
-
- for (String child : children) {
- if (!child.equals("zookeeper")) {
- client.delete().deletingChildrenIfNeeded().forPath('/' +
child);
- }
- }
- }
-
- @AfterClass
- public static void teardownClass() {
- if (client != null) {
- client.close();
- }
- }
-
- /** Tests that a simple {@link ZooKeeperHaServices#close()} does not
delete ZooKeeper paths. */
- @Test
- public void testSimpleClose() throws Exception {
- final String rootPath = "/foo/bar/flink";
- final Configuration configuration = createConfiguration(rootPath);
-
- final TestingBlobStoreService blobStoreService = new
TestingBlobStoreService();
-
- runCleanupTest(configuration, blobStoreService,
ZooKeeperHaServices::close);
-
- assertThat(blobStoreService.isClosed(), is(true));
- assertThat(blobStoreService.isClosedAndCleanedUpAllData(), is(false));
-
- final List<String> children = client.getChildren().forPath(rootPath);
- assertThat(children, is(not(empty())));
- }
-
- /**
- * Tests that the {@link ZooKeeperHaServices} cleans up all paths if it is
closed via {@link
- * ZooKeeperHaServices#closeAndCleanupAllData()}.
- */
- @Test
- public void testSimpleCloseAndCleanupAllData() throws Exception {
- final Configuration configuration =
createConfiguration("/foo/bar/flink");
-
- final TestingBlobStoreService blobStoreService = new
TestingBlobStoreService();
-
- final List<String> initialChildren = client.getChildren().forPath("/");
-
- runCleanupTest(
- configuration, blobStoreService,
ZooKeeperHaServices::closeAndCleanupAllData);
-
- assertThat(blobStoreService.isClosedAndCleanedUpAllData(), is(true));
-
- final List<String> children = client.getChildren().forPath("/");
- assertThat(children, is(equalTo(initialChildren)));
- }
-
- /** Tests that we can only delete the parent znodes as long as they are
empty. */
- @Test
- public void testCloseAndCleanupAllDataWithUncle() throws Exception {
- final String prefix = "/foo/bar";
- final String flinkPath = prefix + "/flink";
- final Configuration configuration = createConfiguration(flinkPath);
-
- final TestingBlobStoreService blobStoreService = new
TestingBlobStoreService();
-
- final String unclePath = prefix + "/foobar";
- client.create().creatingParentContainersIfNeeded().forPath(unclePath);
-
- runCleanupTest(
- configuration, blobStoreService,
ZooKeeperHaServices::closeAndCleanupAllData);
-
- assertThat(blobStoreService.isClosedAndCleanedUpAllData(), is(true));
-
- assertThat(client.checkExists().forPath(flinkPath), is(nullValue()));
- assertThat(client.checkExists().forPath(unclePath),
is(notNullValue()));
- }
-
- /** Tests that the ZooKeeperHaServices cleans up paths for job manager. */
- @Test
- public void testCleanupJobData() throws Exception {
- String rootPath = "/foo/bar/flink";
- final Configuration configuration = createConfiguration(rootPath);
- final String namespace =
configuration.get(HighAvailabilityOptions.HA_CLUSTER_ID);
-
- JobID jobID = new JobID();
- final String path = rootPath + namespace +
ZooKeeperUtils.getJobsPath();
-
- final TestingBlobStoreService blobStoreService = new
TestingBlobStoreService();
-
- runCleanupTestWithJob(
- configuration,
- blobStoreService,
- jobID,
- haServices -> {
- final List<String> childrenBefore =
client.getChildren().forPath(path);
-
- haServices.globalCleanupAsync(jobID,
Executors.directExecutor()).join();
-
- final List<String> childrenAfter =
client.getChildren().forPath(path);
-
- assertThat(childrenBefore, hasItem(jobID.toString()));
- assertThat(childrenAfter, not(hasItem(jobID.toString())));
- });
- }
-
- private static CuratorFramework startCuratorFramework() {
- return CuratorFrameworkFactory.builder()
- .connectString(ZOO_KEEPER_RESOURCE.getConnectString())
- .retryPolicy(new RetryNTimes(50, 100))
- .build();
- }
-
- @Nonnull
- private Configuration createConfiguration(String rootPath) {
- final Configuration configuration = new Configuration();
- configuration.setString(
- HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM,
- ZOO_KEEPER_RESOURCE.getConnectString());
- configuration.setString(HighAvailabilityOptions.HA_ZOOKEEPER_ROOT,
rootPath);
- return configuration;
- }
-
- private void runCleanupTest(
- Configuration configuration,
- TestingBlobStoreService blobStoreService,
- ThrowingConsumer<ZooKeeperHaServices, Exception>
zooKeeperHaServicesConsumer)
- throws Exception {
- runCleanupTestWithJob(
- configuration, blobStoreService, new JobID(),
zooKeeperHaServicesConsumer);
- }
-
- private void runCleanupTestWithJob(
- Configuration configuration,
- TestingBlobStoreService blobStoreService,
- JobID jobId,
- ThrowingConsumer<ZooKeeperHaServices, Exception>
zooKeeperHaServicesConsumer)
- throws Exception {
- try (ZooKeeperHaServices zooKeeperHaServices =
- new ZooKeeperHaServices(
- ZooKeeperUtils.startCuratorFramework(
- configuration,
-
testingFatalErrorHandlerResource.getFatalErrorHandler()),
- Executors.directExecutor(),
- configuration,
- blobStoreService)) {
-
- // create some Zk services to trigger the generation of paths
- final LeaderRetrievalService resourceManagerLeaderRetriever =
- zooKeeperHaServices.getResourceManagerLeaderRetriever();
- final LeaderElectionService resourceManagerLeaderElectionService =
-
zooKeeperHaServices.getResourceManagerLeaderElectionService();
-
- final LeaderRetrievalService jobManagerLeaderRetriever =
- zooKeeperHaServices.getJobManagerLeaderRetriever(jobId);
- final LeaderElectionService jobManagerLeaderElectionService =
-
zooKeeperHaServices.getJobManagerLeaderElectionService(jobId);
-
- final LeaderRetrievalUtils.LeaderConnectionInfoListener
resourceManagerLeaderListener =
- new LeaderRetrievalUtils.LeaderConnectionInfoListener();
- resourceManagerLeaderElectionService.start(
- new TestingContender(
- "unused-resourcemanager-address",
- resourceManagerLeaderElectionService));
-
resourceManagerLeaderRetriever.start(resourceManagerLeaderListener);
-
- final LeaderRetrievalUtils.LeaderConnectionInfoListener
jobManagerLeaderListener =
- new LeaderRetrievalUtils.LeaderConnectionInfoListener();
- jobManagerLeaderElectionService.start(
- new TestingContender(
- "unused-jobmanager-address",
jobManagerLeaderElectionService));
- jobManagerLeaderRetriever.start(jobManagerLeaderListener);
-
- // Make sure that the respective zNodes have been properly created
-
resourceManagerLeaderListener.getLeaderConnectionInfoFuture().join();
- jobManagerLeaderListener.getLeaderConnectionInfoFuture().join();
-
- resourceManagerLeaderRetriever.stop();
- resourceManagerLeaderElectionService.stop();
- jobManagerLeaderRetriever.stop();
- jobManagerLeaderElectionService.stop();
-
- zooKeeperHaServicesConsumer.accept(zooKeeperHaServices);
- }
- }
-
- private static class TestingBlobStoreService implements BlobStoreService {
-
- private boolean closedAndCleanedUpAllData = false;
- private boolean closed = false;
-
- @Override
- public void closeAndCleanupAllData() {
- closedAndCleanedUpAllData = true;
- }
-
- @Override
- public void close() throws IOException {
- closed = true;
- }
-
- @Override
- public boolean put(File localFile, JobID jobId, BlobKey blobKey) {
- return false;
- }
-
- @Override
- public boolean delete(JobID jobId, BlobKey blobKey) {
- return false;
- }
-
- @Override
- public boolean deleteAll(JobID jobId) {
- return false;
- }
-
- @Override
- public boolean get(JobID jobId, BlobKey blobKey, File localFile) {
- return false;
- }
-
- private boolean isClosed() {
- return closed;
- }
-
- private boolean isClosedAndCleanedUpAllData() {
- return closedAndCleanedUpAllData;
- }
- }
-}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalTest.java
index 5978fe78924..31bca72ad67 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalTest.java
@@ -22,7 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.runtime.blob.VoidBlobStore;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
-import org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServices;
+import
org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperMultipleComponentLeaderElectionHaServices;
import org.apache.flink.runtime.jobmaster.JobMaster;
import org.apache.flink.runtime.leaderelection.LeaderElectionService;
import org.apache.flink.runtime.leaderelection.TestingContender;
@@ -83,12 +83,13 @@ public class ZooKeeperLeaderRetrievalTest extends
TestLogger {
HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM,
testingServer.getConnectString());
highAvailabilityServices =
- new ZooKeeperHaServices(
+ new ZooKeeperMultipleComponentLeaderElectionHaServices(
ZooKeeperUtils.startCuratorFramework(
config,
testingFatalErrorHandlerResource.getFatalErrorHandler()),
- EXECUTOR_RESOURCE.getExecutor(),
config,
- new VoidBlobStore());
+ EXECUTOR_RESOURCE.getExecutor(),
+ new VoidBlobStore(),
+
testingFatalErrorHandlerResource.getFatalErrorHandler());
}
@After
