{@link #close()} and {@link #closeAndCleanupAllData()} should be implemented to destroy the @@ -75,17 +75,17 @@ public AbstractHaServices( @Override public LeaderRetrievalService getResourceManagerLeaderRetriever() { - return createLeaderRetrievalService(getLeaderNameForResourceManager()); + return createLeaderRetrievalService(getLeaderPathForResourceManager()); } @Override public LeaderRetrievalService getDispatcherLeaderRetriever() { - return createLeaderRetrievalService(getLeaderNameForDispatcher()); + return createLeaderRetrievalService(getLeaderPathForDispatcher()); } @Override public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID) { - return createLeaderRetrievalService(getLeaderNameForJobManager(jobID)); + return createLeaderRetrievalService(getLeaderPathForJobManager(jobID)); } @Override @@ -96,31 +96,31 @@ public LeaderRetrievalService getJobManagerLeaderRetriever( @Override public LeaderRetrievalService getClusterRestEndpointLeaderRetriever() { - return createLeaderRetrievalService(getLeaderNameForRestServer()); + return createLeaderRetrievalService(getLeaderPathForRestServer()); } @Override public LeaderElectionService getResourceManagerLeaderElectionService() { - return createLeaderElectionService(getLeaderNameForResourceManager()); + return createLeaderElectionService(getLeaderPathForResourceManager()); } @Override public LeaderElectionService getDispatcherLeaderElectionService() { - return createLeaderElectionService(getLeaderNameForDispatcher()); + return createLeaderElectionService(getLeaderPathForDispatcher()); } @Override public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) { - return createLeaderElectionService(getLeaderNameForJobManager(jobID)); + return createLeaderElectionService(getLeaderPathForJobManager(jobID)); } @Override public LeaderElectionService getClusterRestEndpointLeaderElectionService() { - return createLeaderElectionService(getLeaderNameForRestServer()); + return createLeaderElectionService(getLeaderPathForRestServer()); } @Override - public CheckpointRecoveryFactory getCheckpointRecoveryFactory() { + public CheckpointRecoveryFactory getCheckpointRecoveryFactory() throws Exception { return createCheckpointRecoveryFactory(); } @@ -226,7 +226,7 @@ public void closeAndCleanupAllData() throws Exception { * * @return Checkpoint recovery factory */ - protected abstract CheckpointRecoveryFactory createCheckpointRecoveryFactory(); + protected abstract CheckpointRecoveryFactory createCheckpointRecoveryFactory() throws Exception; /** * Create the submitted job graph store for the job manager. @@ -266,7 +266,7 @@ public void closeAndCleanupAllData() throws Exception { * @return Return the ResourceManager leader name. It is ConfigMap name in Kubernetes or child * node path in Zookeeper. */ - protected abstract String getLeaderNameForResourceManager(); + protected abstract String getLeaderPathForResourceManager(); /** * Get the leader name for Dispatcher. @@ -274,7 +274,7 @@ public void closeAndCleanupAllData() throws Exception { * @return Return the Dispatcher leader name. It is ConfigMap name in Kubernetes or child node * path in Zookeeper. */ - protected abstract String getLeaderNameForDispatcher(); + protected abstract String getLeaderPathForDispatcher(); /** * Get the leader name for specific JobManager. @@ -283,7 +283,7 @@ public void closeAndCleanupAllData() throws Exception { * @return Return the JobManager leader name for specified job id. It is ConfigMap name in * Kubernetes or child node path in Zookeeper. */ - protected abstract String getLeaderNameForJobManager(final JobID jobID); + protected abstract String getLeaderPathForJobManager(final JobID jobID); /** * Get the leader name for RestServer. @@ -291,5 +291,5 @@ public void closeAndCleanupAllData() throws Exception { * @return Return the RestServer leader name. It is ConfigMap name in Kubernetes or child node * path in Zookeeper. */ - protected abstract String getLeaderNameForRestServer(); + protected abstract String getLeaderPathForRestServer(); } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java index fdb3fa7dcb5359..650ab971a66ab9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java @@ -158,7 +158,7 @@ default LeaderElectionService getWebMonitorLeaderElectionService() { * * @return Checkpoint recovery factory */ - CheckpointRecoveryFactory getCheckpointRecoveryFactory(); + CheckpointRecoveryFactory getCheckpointRecoveryFactory() throws Exception; /** * Gets the submitted job graph store for the job manager. 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 0f6ebb05eb0e6a..3f134317879367 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 @@ -140,7 +140,7 @@ public static ClientHighAvailabilityServices createClientHAService(Configuration return new StandaloneClientHAServices(webMonitorAddress); case ZOOKEEPER: final CuratorFramework client = ZooKeeperUtils.startCuratorFramework(configuration); - return new ZooKeeperClientHAServices(client, configuration); + return new ZooKeeperClientHAServices(client); case FACTORY_CLASS: return createCustomClientHAServices(configuration); default: diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperClientHAServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperClientHAServices.java index e82d2cb85a0c12..76002c8d667cd4 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperClientHAServices.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperClientHAServices.java @@ -18,7 +18,6 @@ package org.apache.flink.runtime.highavailability.zookeeper; -import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.highavailability.ClientHighAvailabilityServices; import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService; import org.apache.flink.runtime.util.ZooKeeperUtils; @@ -30,21 +29,16 @@ /** ZooKeeper based implementation for {@link ClientHighAvailabilityServices}. */ public class ZooKeeperClientHAServices implements ClientHighAvailabilityServices { - private static final String REST_SERVER_LEADER_PATH = "/rest_server_lock"; - private final CuratorFramework client; - private final Configuration configuration; - public ZooKeeperClientHAServices( - @Nonnull CuratorFramework client, @Nonnull Configuration configuration) { + public ZooKeeperClientHAServices(@Nonnull CuratorFramework client) { this.client = client; - this.configuration = configuration; } @Override public LeaderRetrievalService getClusterRestEndpointLeaderRetriever() { return ZooKeeperUtils.createLeaderRetrievalService( - client, configuration, REST_SERVER_LEADER_PATH); + client, ZooKeeperUtils.getLeaderPathForRestServer()); } @Override 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 index f6ab9e859d630a..f9d43afafe9e38 100644 --- 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 @@ -47,21 +47,37 @@ * *
* /flink - * +/cluster_id_1/resource_manager_lock + * +/cluster_id_1/leader/resource_manager/latch + * | | /connection_info + * | | /dispatcher/latch + * | | /connection_info + * | | /rest_server/latch + * | | /connection_info * | | - * | +/job-id-1/job_manager_lock - * | | /checkpoints/latest - * | | /latest-1 - * | | /latest-2 * | | - * | +/job-id-2/job_manager_lock + * | +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/resource_manager_lock - * | - * +/job-id-1/job_manager_lock - * |/checkpoints/latest - * | /latest-1 - * |/persisted_job_graph + * +/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 ** *
The root path "/flink" is configurable via the option {@link
@@ -81,14 +97,6 @@
*/
public class ZooKeeperHaServices extends AbstractHaServices {
- private static final String RESOURCE_MANAGER_LEADER_PATH = "/resource_manager_lock";
-
- private static final String DISPATCHER_LEADER_PATH = "/dispatcher_lock";
-
- private static final String JOB_MANAGER_LEADER_PATH = "/job_manager_lock";
-
- private static final String REST_SERVER_LEADER_PATH = "/rest_server_lock";
-
// ------------------------------------------------------------------------
/** The ZooKeeper client to use. */
@@ -104,8 +112,11 @@ public ZooKeeperHaServices(
}
@Override
- public CheckpointRecoveryFactory createCheckpointRecoveryFactory() {
- return new ZooKeeperCheckpointRecoveryFactory(client, configuration, ioExecutor);
+ public CheckpointRecoveryFactory createCheckpointRecoveryFactory() throws Exception {
+ return new ZooKeeperCheckpointRecoveryFactory(
+ ZooKeeperUtils.useNamespaceAndEnsurePath(client, ZooKeeperUtils.getJobsPath()),
+ configuration,
+ ioExecutor);
}
@Override
@@ -119,13 +130,13 @@ public RunningJobsRegistry createRunningJobsRegistry() {
}
@Override
- protected LeaderElectionService createLeaderElectionService(String leaderName) {
- return ZooKeeperUtils.createLeaderElectionService(client, configuration, leaderName);
+ protected LeaderElectionService createLeaderElectionService(String leaderPath) {
+ return ZooKeeperUtils.createLeaderElectionService(client, leaderPath);
}
@Override
- protected LeaderRetrievalService createLeaderRetrievalService(String leaderName) {
- return ZooKeeperUtils.createLeaderRetrievalService(client, configuration, leaderName);
+ protected LeaderRetrievalService createLeaderRetrievalService(String leaderPath) {
+ return ZooKeeperUtils.createLeaderRetrievalService(client, leaderPath);
}
@Override
@@ -139,22 +150,22 @@ public void internalCleanup() throws Exception {
}
@Override
- protected String getLeaderNameForResourceManager() {
- return RESOURCE_MANAGER_LEADER_PATH;
+ protected String getLeaderPathForResourceManager() {
+ return ZooKeeperUtils.getLeaderPathForResourceManager();
}
@Override
- protected String getLeaderNameForDispatcher() {
- return DISPATCHER_LEADER_PATH;
+ protected String getLeaderPathForDispatcher() {
+ return ZooKeeperUtils.getLeaderPathForDispatcher();
}
- public String getLeaderNameForJobManager(final JobID jobID) {
- return "/" + jobID + JOB_MANAGER_LEADER_PATH;
+ public String getLeaderPathForJobManager(final JobID jobID) {
+ return ZooKeeperUtils.getLeaderPathForJobManager(jobID);
}
@Override
- protected String getLeaderNameForRestServer() {
- return REST_SERVER_LEADER_PATH;
+ protected String getLeaderPathForRestServer() {
+ return ZooKeeperUtils.getLeaderPathForRestServer();
}
// ------------------------------------------------------------------------
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
index 855ebca247a38c..522f36eb93b856 100644
--- 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
@@ -18,7 +18,9 @@
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.shaded.curator4.org.apache.curator.framework.CuratorFramework;
@@ -68,7 +70,7 @@ public class ZooKeeperLeaderElectionDriver
private final NodeCache cache;
/** ZooKeeper path of the node which stores the current leader information. */
- private final String leaderPath;
+ private final String connectionInformationPath;
private final ConnectionStateListener listener =
(client, newState) -> handleStateChange(newState);
@@ -85,29 +87,27 @@ public class ZooKeeperLeaderElectionDriver
* Creates a ZooKeeperLeaderElectionDriver object.
*
* @param client Client which is connected to the ZooKeeper quorum
- * @param latchPath ZooKeeper node path for the leader election latch
- * @param leaderPath ZooKeeper node path for the node which stores the current leader
- * information
+ * @param path ZooKeeper node path for the leader election
* @param leaderElectionEventHandler Event handler for processing leader change events
* @param fatalErrorHandler Fatal error handler
* @param leaderContenderDescription Leader contender description
*/
public ZooKeeperLeaderElectionDriver(
CuratorFramework client,
- String latchPath,
- String leaderPath,
+ String path,
LeaderElectionEventHandler leaderElectionEventHandler,
FatalErrorHandler fatalErrorHandler,
String leaderContenderDescription)
throws Exception {
+ checkNotNull(path);
this.client = checkNotNull(client);
- this.leaderPath = checkNotNull(leaderPath);
+ this.connectionInformationPath = ZooKeeperUtils.generateConnectionInformationPath(path);
this.leaderElectionEventHandler = checkNotNull(leaderElectionEventHandler);
this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
this.leaderContenderDescription = checkNotNull(leaderContenderDescription);
- leaderLatch = new LeaderLatch(client, checkNotNull(latchPath));
- cache = new NodeCache(client, leaderPath);
+ leaderLatch = new LeaderLatch(client, ZooKeeperUtils.generateLeaderLatchPath(path));
+ cache = new NodeCache(client, connectionInformationPath);
client.getUnhandledErrorListenable().addListener(this);
@@ -220,7 +220,7 @@ public void writeLeaderInformation(LeaderInformation leaderInformation) {
boolean dataWritten = false;
while (!dataWritten && leaderLatch.hasLeadership()) {
- Stat stat = client.checkExists().forPath(leaderPath);
+ Stat stat = client.checkExists().forPath(connectionInformationPath);
if (stat != null) {
long owner = stat.getEphemeralOwner();
@@ -228,7 +228,7 @@ public void writeLeaderInformation(LeaderInformation leaderInformation) {
if (owner == sessionID) {
try {
- client.setData().forPath(leaderPath, baos.toByteArray());
+ client.setData().forPath(connectionInformationPath, baos.toByteArray());
dataWritten = true;
} catch (KeeperException.NoNodeException noNode) {
@@ -236,7 +236,7 @@ public void writeLeaderInformation(LeaderInformation leaderInformation) {
}
} else {
try {
- client.delete().forPath(leaderPath);
+ client.delete().forPath(connectionInformationPath);
} catch (KeeperException.NoNodeException noNode) {
// node was deleted in the meantime --> try again
}
@@ -246,7 +246,7 @@ public void writeLeaderInformation(LeaderInformation leaderInformation) {
client.create()
.creatingParentsIfNeeded()
.withMode(CreateMode.EPHEMERAL)
- .forPath(leaderPath, baos.toByteArray());
+ .forPath(connectionInformationPath, baos.toByteArray());
dataWritten = true;
} catch (KeeperException.NodeExistsException nodeExists) {
@@ -301,6 +301,15 @@ public void unhandledError(String message, Throwable e) {
@Override
public String toString() {
- return "ZooKeeperLeaderElectionDriver{" + "leaderPath='" + leaderPath + '\'' + '}';
+ return "ZooKeeperLeaderElectionDriver{"
+ + "leaderPath='"
+ + 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
index a408882ef43e05..eef7d0cb3547e0 100644
--- 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
@@ -27,15 +27,11 @@ public class ZooKeeperLeaderElectionDriverFactory implements LeaderElectionDrive
private final CuratorFramework client;
- private final String latchPath;
+ private final String path;
- private final String leaderPath;
-
- public ZooKeeperLeaderElectionDriverFactory(
- CuratorFramework client, String latchPath, String leaderPath) {
+ public ZooKeeperLeaderElectionDriverFactory(CuratorFramework client, String path) {
this.client = client;
- this.latchPath = latchPath;
- this.leaderPath = leaderPath;
+ this.path = path;
}
@Override
@@ -45,11 +41,6 @@ public ZooKeeperLeaderElectionDriver createLeaderElectionDriver(
String leaderContenderDescription)
throws Exception {
return new ZooKeeperLeaderElectionDriver(
- client,
- latchPath,
- leaderPath,
- leaderEventHandler,
- fatalErrorHandler,
- leaderContenderDescription);
+ client, path, leaderEventHandler, fatalErrorHandler, leaderContenderDescription);
}
}
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 f447a17e1ce4e7..c1dc1ff21405a4 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
@@ -18,9 +18,11 @@
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.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.util.ZooKeeperUtils;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFramework;
@@ -57,7 +59,7 @@ public class ZooKeeperLeaderRetrievalDriver
/** Curator recipe to watch changes of a specific ZooKeeper node. */
private final NodeCache cache;
- private final String retrievalPath;
+ private final String connectionInformationPath;
private final ConnectionStateListener connectionStateListener =
(client, newState) -> handleStateChange(newState);
@@ -72,19 +74,19 @@ public class ZooKeeperLeaderRetrievalDriver
* Creates a leader retrieval service which uses ZooKeeper to retrieve the leader information.
*
* @param client Client which constitutes the connection to the ZooKeeper quorum
- * @param retrievalPath Path of the ZooKeeper node which contains the leader information
+ * @param path Path of the ZooKeeper node which contains the leader information
* @param leaderRetrievalEventHandler Handler to notify the leader changes.
* @param fatalErrorHandler Fatal error handler
*/
public ZooKeeperLeaderRetrievalDriver(
CuratorFramework client,
- String retrievalPath,
+ String path,
LeaderRetrievalEventHandler leaderRetrievalEventHandler,
FatalErrorHandler fatalErrorHandler)
throws Exception {
this.client = checkNotNull(client, "CuratorFramework client");
- this.cache = new NodeCache(client, retrievalPath);
- this.retrievalPath = checkNotNull(retrievalPath);
+ this.connectionInformationPath = ZooKeeperUtils.generateConnectionInformationPath(path);
+ this.cache = new NodeCache(client, connectionInformationPath);
this.leaderRetrievalEventHandler = checkNotNull(leaderRetrievalEventHandler);
this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
@@ -188,6 +190,15 @@ public void unhandledError(String s, Throwable throwable) {
@Override
public String toString() {
- return "ZookeeperLeaderRetrievalDriver{" + "retrievalPath='" + retrievalPath + '\'' + '}';
+ return "ZookeeperLeaderRetrievalDriver{"
+ + "retrievalPath='"
+ + connectionInformationPath
+ + '\''
+ + '}';
+ }
+
+ @VisibleForTesting
+ public String getConnectionInformationPath() {
+ return connectionInformationPath;
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java
index 0af2248340ffed..4164942b93e580 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java
@@ -47,7 +47,6 @@
import org.apache.flink.runtime.persistence.RetrievableStateStorageHelper;
import org.apache.flink.runtime.persistence.filesystem.FileSystemStateStorageHelper;
import org.apache.flink.runtime.zookeeper.ZooKeeperStateHandleStore;
-import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFramework;
import org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFrameworkFactory;
@@ -62,12 +61,17 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nonnull;
+
import java.io.IOException;
import java.io.Serializable;
import java.util.List;
import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
/** Class containing helper functions to interact with ZooKeeper. */
public class ZooKeeperUtils {
@@ -80,6 +84,56 @@ public class ZooKeeperUtils {
/** The prefix of the completed checkpoint file. */
public static final String HA_STORAGE_COMPLETED_CHECKPOINT = "completedCheckpoint";
+ private static final String RESOURCE_MANAGER_LEADER = "/resource_manager";
+
+ private static final String DISPATCHER_LEADER = "/dispatcher";
+
+ private static final String LEADER_NODE = "/leader";
+
+ private static final String REST_SERVER_LEADER = "/rest_server";
+
+ public static String getLeaderPathForResourceManager() {
+ return getLeaderPath(RESOURCE_MANAGER_LEADER);
+ }
+
+ public static String getLeaderPathForDispatcher() {
+ return getLeaderPath(DISPATCHER_LEADER);
+ }
+
+ public static String getLeaderPathForRestServer() {
+ return getLeaderPath(REST_SERVER_LEADER);
+ }
+
+ public static String getLeaderPathForJobManager(JobID jobId) {
+ return generateZookeeperPath(
+ generateZookeeperPath(getJobsPath(), getPathForJob(jobId)), LEADER_NODE);
+ }
+
+ public static String getJobsPath() {
+ return "/jobs";
+ }
+
+ private static String getCheckpointsPath() {
+ return "/checkpoints";
+ }
+
+ public static String getCheckpointIdCounterPath() {
+ return "/checkpoint_id_counter";
+ }
+
+ private static String getLeaderPath(String suffix) {
+ return generateZookeeperPath(LEADER_NODE, suffix);
+ }
+
+ @Nonnull
+ public static String generateConnectionInformationPath(String path) {
+ return generateZookeeperPath(path, "connection_info");
+ }
+
+ public static String generateLeaderLatchPath(String path) {
+ return generateZookeeperPath(path, "latch");
+ }
+
/**
* Starts a {@link CuratorFramework} instance and connects it to the given ZooKeeper quorum.
*
@@ -87,7 +141,7 @@ public class ZooKeeperUtils {
* @return {@link CuratorFramework} instance
*/
public static CuratorFramework startCuratorFramework(Configuration configuration) {
- Preconditions.checkNotNull(configuration, "configuration");
+ checkNotNull(configuration, "configuration");
String zkQuorum = configuration.getValue(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM);
if (zkQuorum == null || StringUtils.isBlank(zkQuorum)) {
@@ -123,7 +177,7 @@ public static CuratorFramework startCuratorFramework(Configuration configuration
if (disableSaslClient && aclMode == ZkClientACLMode.CREATOR) {
String errorMessage =
"Cannot set ACL role to "
- + aclMode
+ + ZkClientACLMode.CREATOR
+ " since SASL authentication is "
+ "disabled through the "
+ SecurityOptions.ZOOKEEPER_SASL_DISABLE.key()
@@ -193,12 +247,11 @@ public static String getZooKeeperEnsemble(Configuration flinkConf)
* ZooKeeperLeaderRetrievalDriver}.
*
* @param client The {@link CuratorFramework} ZooKeeper client to use
- * @param configuration {@link Configuration} object containing the configuration values
* @return {@link DefaultLeaderRetrievalService} instance.
*/
public static DefaultLeaderRetrievalService createLeaderRetrievalService(
- final CuratorFramework client, final Configuration configuration) {
- return createLeaderRetrievalService(client, configuration, "");
+ final CuratorFramework client) {
+ return createLeaderRetrievalService(client, "");
}
/**
@@ -206,46 +259,35 @@ public static DefaultLeaderRetrievalService createLeaderRetrievalService(
* ZooKeeperLeaderRetrievalDriver}.
*
* @param client The {@link CuratorFramework} ZooKeeper client to use
- * @param configuration {@link Configuration} object containing the configuration values
- * @param pathSuffix The path suffix which we want to append
+ * @param path The path for the leader retrieval
* @return {@link DefaultLeaderRetrievalService} instance.
*/
public static DefaultLeaderRetrievalService createLeaderRetrievalService(
- final CuratorFramework client,
- final Configuration configuration,
- final String pathSuffix) {
- return new DefaultLeaderRetrievalService(
- createLeaderRetrievalDriverFactory(client, configuration, pathSuffix));
+ final CuratorFramework client, final String path) {
+ return new DefaultLeaderRetrievalService(createLeaderRetrievalDriverFactory(client, path));
}
/**
* Creates a {@link LeaderRetrievalDriverFactory} implemented by ZooKeeper.
*
* @param client The {@link CuratorFramework} ZooKeeper client to use
- * @param configuration {@link Configuration} object containing the configuration values
* @return {@link LeaderRetrievalDriverFactory} instance.
*/
public static ZooKeeperLeaderRetrievalDriverFactory createLeaderRetrievalDriverFactory(
- final CuratorFramework client, final Configuration configuration) {
- return createLeaderRetrievalDriverFactory(client, configuration, "");
+ final CuratorFramework client) {
+ return createLeaderRetrievalDriverFactory(client, "");
}
/**
* Creates a {@link LeaderRetrievalDriverFactory} implemented by ZooKeeper.
*
* @param client The {@link CuratorFramework} ZooKeeper client to use
- * @param configuration {@link Configuration} object containing the configuration values
- * @param pathSuffix The path suffix which we want to append
+ * @param path The path for the leader zNode
* @return {@link LeaderRetrievalDriverFactory} instance.
*/
public static ZooKeeperLeaderRetrievalDriverFactory createLeaderRetrievalDriverFactory(
- final CuratorFramework client,
- final Configuration configuration,
- final String pathSuffix) {
- final String leaderPath =
- configuration.getString(HighAvailabilityOptions.HA_ZOOKEEPER_LEADER_PATH)
- + pathSuffix;
- return new ZooKeeperLeaderRetrievalDriverFactory(client, leaderPath);
+ final CuratorFramework client, final String path) {
+ return new ZooKeeperLeaderRetrievalDriverFactory(client, path);
}
/**
@@ -253,13 +295,12 @@ public static ZooKeeperLeaderRetrievalDriverFactory createLeaderRetrievalDriverF
* ZooKeeperLeaderElectionDriver}.
*
* @param client The {@link CuratorFramework} ZooKeeper client to use
- * @param configuration {@link Configuration} object containing the configuration values
* @return {@link DefaultLeaderElectionService} instance.
*/
public static DefaultLeaderElectionService createLeaderElectionService(
- CuratorFramework client, Configuration configuration) {
+ CuratorFramework client) {
- return createLeaderElectionService(client, configuration, "");
+ return createLeaderElectionService(client, "");
}
/**
@@ -267,50 +308,35 @@ public static DefaultLeaderElectionService createLeaderElectionService(
* ZooKeeperLeaderElectionDriver}.
*
* @param client The {@link CuratorFramework} ZooKeeper client to use
- * @param configuration {@link Configuration} object containing the configuration values
- * @param pathSuffix The path suffix which we want to append
+ * @param path The path for the leader election
* @return {@link DefaultLeaderElectionService} instance.
*/
public static DefaultLeaderElectionService createLeaderElectionService(
- final CuratorFramework client,
- final Configuration configuration,
- final String pathSuffix) {
- return new DefaultLeaderElectionService(
- createLeaderElectionDriverFactory(client, configuration, pathSuffix));
+ final CuratorFramework client, final String path) {
+ return new DefaultLeaderElectionService(createLeaderElectionDriverFactory(client, path));
}
/**
* Creates a {@link LeaderElectionDriverFactory} implemented by ZooKeeper.
*
* @param client The {@link CuratorFramework} ZooKeeper client to use
- * @param configuration {@link Configuration} object containing the configuration values
* @return {@link LeaderElectionDriverFactory} instance.
*/
public static ZooKeeperLeaderElectionDriverFactory createLeaderElectionDriverFactory(
- final CuratorFramework client, final Configuration configuration) {
- return createLeaderElectionDriverFactory(client, configuration, "");
+ final CuratorFramework client) {
+ return createLeaderElectionDriverFactory(client, "");
}
/**
* Creates a {@link LeaderElectionDriverFactory} implemented by ZooKeeper.
*
* @param client The {@link CuratorFramework} ZooKeeper client to use
- * @param configuration {@link Configuration} object containing the configuration values
- * @param pathSuffix The path suffix which we want to append
+ * @param path The path suffix which we want to append
* @return {@link LeaderElectionDriverFactory} instance.
*/
public static ZooKeeperLeaderElectionDriverFactory createLeaderElectionDriverFactory(
- final CuratorFramework client,
- final Configuration configuration,
- final String pathSuffix) {
- final String latchPath =
- configuration.getString(HighAvailabilityOptions.HA_ZOOKEEPER_LATCH_PATH)
- + pathSuffix;
- final String leaderPath =
- configuration.getString(HighAvailabilityOptions.HA_ZOOKEEPER_LEADER_PATH)
- + pathSuffix;
-
- return new ZooKeeperLeaderElectionDriverFactory(client, latchPath, leaderPath);
+ final CuratorFramework client, final String path) {
+ return new ZooKeeperLeaderElectionDriverFactory(client, path);
}
/**
@@ -359,7 +385,6 @@ public static JobGraphStore createJobGraphs(
*
* @param client The {@link CuratorFramework} ZooKeeper client to use
* @param configuration {@link Configuration} object
- * @param jobId ID of job to create the instance for
* @param maxNumberOfCheckpointsToRetain The maximum number of checkpoints to retain
* @param executor to run ZooKeeper callbacks
* @return {@link DefaultCompletedCheckpointStore} instance
@@ -368,23 +393,17 @@ public static JobGraphStore createJobGraphs(
public static CompletedCheckpointStore createCompletedCheckpoints(
CuratorFramework client,
Configuration configuration,
- JobID jobId,
int maxNumberOfCheckpointsToRetain,
Executor executor)
throws Exception {
checkNotNull(configuration, "Configuration");
- String checkpointsPath =
- configuration.getString(HighAvailabilityOptions.HA_ZOOKEEPER_CHECKPOINTS_PATH);
-
RetrievableStateStorageHelper