You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2015/10/23 15:34:53 UTC
[3/5] flink git commit: [FLINK-2893] [runtime] Consistent naming of
recovery config parameters
[FLINK-2893] [runtime] Consistent naming of recovery config parameters
Rename config key prefix from 'ha.zookeeper' to 'recovery.zookeeper'
Rename config key from 'state.backend.fs.dir.recovery' => 'state.backend.fs.recoverydir'
Move ZooKeeper file system state backend configuration keys
This closes #1286
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ab2895fa
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ab2895fa
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ab2895fa
Branch: refs/heads/master
Commit: ab2895fa988018300395557d110de23a3a2166c9
Parents: 3c8a658
Author: Ufuk Celebi <uc...@apache.org>
Authored: Thu Oct 22 12:58:26 2015 +0200
Committer: Fabian Hueske <fh...@apache.org>
Committed: Fri Oct 23 13:03:02 2015 +0200
----------------------------------------------------------------------
docs/setup/config.md | 18 +++++------
docs/setup/jobmanager_high_availability.md | 6 ++--
.../flink/configuration/ConfigConstants.java | 34 ++++++++++----------
flink-dist/src/main/resources/flink-conf.yaml | 2 +-
.../apache/flink/runtime/blob/BlobServer.java | 2 +-
.../flink/runtime/blob/FileSystemBlobStore.java | 4 +--
.../flink/runtime/util/ZooKeeperUtils.java | 4 +--
.../flink/runtime/blob/BlobRecoveryITCase.java | 2 +-
.../BlobLibraryCacheRecoveryITCase.java | 2 +-
.../runtime/testutils/JobManagerProcess.java | 2 +-
.../runtime/testutils/TaskManagerProcess.java | 2 +-
.../runtime/testutils/ZooKeeperTestUtils.java | 2 +-
.../flink/test/recovery/ChaosMonkeyITCase.java | 2 +-
.../ZooKeeperLeaderElectionITCase.java | 4 +--
.../flink/yarn/YARNHighAvailabilityITCase.java | 4 +--
15 files changed, 45 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/docs/setup/config.md
----------------------------------------------------------------------
diff --git a/docs/setup/config.md b/docs/setup/config.md
index 0e7b2ee..447a329 100644
--- a/docs/setup/config.md
+++ b/docs/setup/config.md
@@ -390,23 +390,23 @@ Flink supports the 'standalone' mode where only a single JobManager runs and no
The high availability mode 'zookeeper' supports the execution of multiple JobManagers and JobManager state checkpointing.
Among the group of JobManagers, ZooKeeper elects one of them as the leader which is responsible for the cluster execution.
In case of a JobManager failure, a standby JobManager will be elected as the new leader and is given the last checkpointed JobManager state.
-In order to use the 'zookeeper' mode, it is mandatory to also define the `ha.zookeeper.quorum` configuration value.
+In order to use the 'zookeeper' mode, it is mandatory to also define the `recovery.zookeeper.quorum` configuration value.
-- `ha.zookeeper.quorum`: Defines the ZooKeeper quorum URL which is used to connet to the ZooKeeper cluster when the 'zookeeper' recovery mode is selected
+- `recovery.zookeeper.quorum`: Defines the ZooKeeper quorum URL which is used to connet to the ZooKeeper cluster when the 'zookeeper' recovery mode is selected
-- `ha.zookeeper.dir`: (Default '/flink') Defines the root dir under which the ZooKeeper recovery mode will create znodes.
+- `recovery.zookeeper.path.root`: (Default '/flink') Defines the root dir under which the ZooKeeper recovery mode will create znodes.
-- `ha.zookeeper.dir.latch`: (Default '/leaderlatch') Defines the znode of the leader latch which is used to elect the leader.
+- `recovery.zookeeper.path.latch`: (Default '/leaderlatch') Defines the znode of the leader latch which is used to elect the leader.
-- `ha.zookeeper.dir.leader`: (Default '/leader') Defines the znode of the leader which contains the URL to the leader and the current leader session ID
+- `recovery.zookeeper.path.leader`: (Default '/leader') Defines the znode of the leader which contains the URL to the leader and the current leader session ID
-- `ha.zookeeper.client.session-timeout`: (Default '60000') Defines the session timeout for the ZooKeeper session in ms.
+- `recovery.zookeeper.client.session-timeout`: (Default '60000') Defines the session timeout for the ZooKeeper session in ms.
-- `ha.zookeeper.client.connection-timeout`: (Default '15000') Defines the connection timeout for ZooKeeper in ms.
+- `recovery.zookeeper.client.connection-timeout`: (Default '15000') Defines the connection timeout for ZooKeeper in ms.
-- `ha.zookeeper.client.retry-wait`: (Default '5000') Defines the pause between consecutive retries in ms.
+- `recovery.zookeeper.client.retry-wait`: (Default '5000') Defines the pause between consecutive retries in ms.
-- `ha.zookeeper.client.max-retry-attempts`: (Default '3') Defines the number of connection retries before the client gives up.
+- `recovery.zookeeper.client.max-retry-attempts`: (Default '3') Defines the number of connection retries before the client gives up.
## Background
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/docs/setup/jobmanager_high_availability.md
----------------------------------------------------------------------
diff --git a/docs/setup/jobmanager_high_availability.md b/docs/setup/jobmanager_high_availability.md
index 50379ea..55b15ad 100644
--- a/docs/setup/jobmanager_high_availability.md
+++ b/docs/setup/jobmanager_high_availability.md
@@ -50,13 +50,13 @@ In high availabliity mode, all Flink components try to connect to a JobManager v
- **ZooKeeper quorum** (required): A *ZooKeeper quorum* is a replicated group of ZooKeeper servers, which provide the distributed coordination service.
- <pre>ha.zookeeper.quorum: address1:2181[,...],addressX:2181</pre>
+ <pre>recovery.zookeeper.quorum: address1:2181[,...],addressX:2181</pre>
Each *addressX:port* refers to a ZooKeeper server, which is reachable by Flink at the given address and port.
- The following configuration keys are optional:
- - `ha.zookeeper.dir: /flink [default]`: ZooKeeper directory to use for coordination
+ - `recovery.zookeeper.path.root: /flink [default]`: ZooKeeper directory to use for coordination
- TODO Add client configuration keys
## Starting an HA-cluster
@@ -93,7 +93,7 @@ The script `bin/start-zookeeper-quorum.sh` will start a ZooKeeper server on each
<pre>
recovery.mode: zookeeper
-ha.zookeeper.quorum: localhost</pre>
+recovery.zookeeper.quorum: localhost</pre>
2. **Configure masters** in `conf/masters`:
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index fc2087a..b64939e 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -410,12 +410,6 @@ public final class ConfigConstants {
*/
public static final String STATE_BACKEND = "state.backend";
- /**
- * File system state backend base path for recoverable state handles. Recovery state is written
- * to this path and the file state handles are persisted for recovery.
- */
- public static final String STATE_BACKEND_FS_RECOVERY_PATH = "state.backend.fs.dir.recovery";
-
// ----------------------------- Miscellaneous ----------------------------
/**
@@ -433,31 +427,37 @@ public final class ConfigConstants {
// --------------------------- ZooKeeper ----------------------------------
/** ZooKeeper servers. */
- public static final String ZOOKEEPER_QUORUM_KEY = "ha.zookeeper.quorum";
+ public static final String ZOOKEEPER_QUORUM_KEY = "recovery.zookeeper.quorum";
+
+ /**
+ * File system state backend base path for recoverable state handles. Recovery state is written
+ * to this path and the file state handles are persisted for recovery.
+ */
+ public static final String ZOOKEEPER_RECOVERY_PATH = "recovery.zookeeper.storageDir";
/** ZooKeeper root path. */
- public static final String ZOOKEEPER_DIR_KEY = "ha.zookeeper.dir";
+ public static final String ZOOKEEPER_DIR_KEY = "recovery.zookeeper.path.root";
- public static final String ZOOKEEPER_LATCH_PATH = "ha.zookeeper.dir.latch";
+ public static final String ZOOKEEPER_LATCH_PATH = "recovery.zookeeper.path.latch";
- public static final String ZOOKEEPER_LEADER_PATH = "ha.zookeeper.dir.leader";
+ public static final String ZOOKEEPER_LEADER_PATH = "recovery.zookeeper.path.leader";
/** ZooKeeper root path (ZNode) for job graphs. */
- public static final String ZOOKEEPER_JOBGRAPHS_PATH = "ha.zookeeper.dir.jobgraphs";
+ public static final String ZOOKEEPER_JOBGRAPHS_PATH = "recovery.zookeeper.path.jobgraphs";
/** ZooKeeper root path (ZNode) for completed checkpoints. */
- public static final String ZOOKEEPER_CHECKPOINTS_PATH = "ha.zookeeper.dir.checkpoints";
+ public static final String ZOOKEEPER_CHECKPOINTS_PATH = "recovery.zookeeper.path.checkpoints";
/** ZooKeeper root path (ZNode) for checkpoint counters. */
- public static final String ZOOKEEPER_CHECKPOINT_COUNTER_PATH = "ha.zookeeper.dir.checkpoint-counter";
+ public static final String ZOOKEEPER_CHECKPOINT_COUNTER_PATH = "recovery.zookeeper.path.checkpoint-counter";
- public static final String ZOOKEEPER_SESSION_TIMEOUT = "ha.zookeeper.client.session-timeout";
+ public static final String ZOOKEEPER_SESSION_TIMEOUT = "recovery.zookeeper.client.session-timeout";
- public static final String ZOOKEEPER_CONNECTION_TIMEOUT = "ha.zookeeper.client.connection-timeout";
+ public static final String ZOOKEEPER_CONNECTION_TIMEOUT = "recovery.zookeeper.client.connection-timeout";
- public static final String ZOOKEEPER_RETRY_WAIT = "ha.zookeeper.client.retry-wait";
+ public static final String ZOOKEEPER_RETRY_WAIT = "recovery.zookeeper.client.retry-wait";
- public static final String ZOOKEEPER_MAX_RETRY_ATTEMPTS = "ha.zookeeper.client.max-retry-attempts";
+ public static final String ZOOKEEPER_MAX_RETRY_ATTEMPTS = "recovery.zookeeper.client.max-retry-attempts";
// ------------------------------------------------------------------------
// Default Values
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-dist/src/main/resources/flink-conf.yaml
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/resources/flink-conf.yaml b/flink-dist/src/main/resources/flink-conf.yaml
index 157e34a..5928d93 100644
--- a/flink-dist/src/main/resources/flink-conf.yaml
+++ b/flink-dist/src/main/resources/flink-conf.yaml
@@ -137,4 +137,4 @@ webclient.port: 8080
#
# recovery.mode: zookeeper
#
-# ha.zookeeper.quorum: localhost
+# recovery.zookeeper.quorum: localhost
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
index d0bed8c..851cff4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
@@ -100,7 +100,7 @@ public class BlobServer extends Thread implements BlobService {
// Recovery. Check that everything has been setup correctly. This is not clean, but it's
// better to resolve this with some upcoming changes to the state backend setup.
else if (config.containsKey(ConfigConstants.STATE_BACKEND) &&
- config.containsKey(ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH)) {
+ config.containsKey(ConfigConstants.ZOOKEEPER_RECOVERY_PATH)) {
this.blobStore = new FileSystemBlobStore(config);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java
index 4351eb1..4c63873 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java
@@ -51,12 +51,12 @@ class FileSystemBlobStore implements BlobStore {
FileSystemBlobStore(Configuration config) throws IOException {
String stateBackendBasePath = config.getString(
- ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH, "");
+ ConfigConstants.ZOOKEEPER_RECOVERY_PATH, "");
if (stateBackendBasePath.equals("")) {
throw new IllegalConfigurationException(String.format("Missing configuration for " +
"file system state backend recovery path. Please specify via " +
- "'%s' key.", ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH));
+ "'%s' key.", ConfigConstants.ZOOKEEPER_RECOVERY_PATH));
}
stateBackendBasePath += "/blob";
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java
----------------------------------------------------------------------
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 a32fc65..79bd28b 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
@@ -276,11 +276,11 @@ public class ZooKeeperUtils {
String prefix) throws IOException {
String rootPath = configuration.getString(
- ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH, "");
+ ConfigConstants.ZOOKEEPER_RECOVERY_PATH, "");
if (rootPath.equals("")) {
throw new IllegalConfigurationException("Missing recovery path. Specify via " +
- "configuration key '" + ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH + "'.");
+ "configuration key '" + ConfigConstants.ZOOKEEPER_RECOVERY_PATH + "'.");
} else {
return new FileSystemStateStorageHelper<T>(rootPath, prefix);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobRecoveryITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobRecoveryITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobRecoveryITCase.java
index 0e324a8..2eff4fc 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobRecoveryITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobRecoveryITCase.java
@@ -70,7 +70,7 @@ public class BlobRecoveryITCase {
Configuration config = new Configuration();
config.setString(ConfigConstants.RECOVERY_MODE, "ZOOKEEPER");
config.setString(ConfigConstants.STATE_BACKEND, "FILESYSTEM");
- config.setString(ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH, recoveryDir.getPath());
+ config.setString(ConfigConstants.ZOOKEEPER_RECOVERY_PATH, recoveryDir.getPath());
for (int i = 0; i < server.length; i++) {
server[i] = new BlobServer(config);
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java
index ea4195c..6068322 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java
@@ -65,7 +65,7 @@ public class BlobLibraryCacheRecoveryITCase {
Configuration config = new Configuration();
config.setString(ConfigConstants.RECOVERY_MODE, "ZOOKEEPER");
config.setString(ConfigConstants.STATE_BACKEND, "FILESYSTEM");
- config.setString(ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH, temporaryFolder.getRoot().getAbsolutePath());
+ config.setString(ConfigConstants.ZOOKEEPER_RECOVERY_PATH, temporaryFolder.getRoot().getAbsolutePath());
for (int i = 0; i < server.length; i++) {
server[i] = new BlobServer(config);
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/JobManagerProcess.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/JobManagerProcess.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/JobManagerProcess.java
index 85b768d..0641493 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/JobManagerProcess.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/JobManagerProcess.java
@@ -197,7 +197,7 @@ public class JobManagerProcess extends TestJvmProcess {
* <code>--port PORT</code>.
*
* <p>Other arguments are parsed to a {@link Configuration} and passed to the
- * JobManager, for instance: <code>--recovery.mode ZOOKEEPER --ha.zookeeper.quorum
+ * JobManager, for instance: <code>--recovery.mode ZOOKEEPER --recovery.zookeeper.quorum
* "xyz:123:456"</code>.
*/
public static void main(String[] args) {
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TaskManagerProcess.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TaskManagerProcess.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TaskManagerProcess.java
index f683c55..86449a8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TaskManagerProcess.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TaskManagerProcess.java
@@ -99,7 +99,7 @@ public class TaskManagerProcess extends TestJvmProcess {
* and streaming jobs).
*
* <p>All arguments are parsed to a {@link Configuration} and passed to the Taskmanager,
- * for instance: <code>--recovery.mode ZOOKEEPER --ha.zookeeper.quorum "xyz:123:456"</code>.
+ * for instance: <code>--recovery.mode ZOOKEEPER --recovery.zookeeper.quorum "xyz:123:456"</code>.
*/
public static void main(String[] args) throws Exception {
try {
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java
index a65ec01..e0f8625 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java
@@ -81,7 +81,7 @@ public class ZooKeeperTestUtils {
// File system state backend
config.setString(ConfigConstants.STATE_BACKEND, "FILESYSTEM");
config.setString(FsStateBackendFactory.CHECKPOINT_DIRECTORY_URI_CONF_KEY, fsStateHandlePath + "/checkpoints");
- config.setString(ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH, fsStateHandlePath + "/recovery");
+ config.setString(ConfigConstants.ZOOKEEPER_RECOVERY_PATH, fsStateHandlePath + "/recovery");
// Akka failure detection and execution retries
config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms");
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-tests/src/test/java/org/apache/flink/test/recovery/ChaosMonkeyITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/ChaosMonkeyITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/ChaosMonkeyITCase.java
index f15644e..f536418 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/ChaosMonkeyITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/ChaosMonkeyITCase.java
@@ -556,7 +556,7 @@ public class ChaosMonkeyITCase {
0, files.length);
}
- File fsRecovery = new File(config.getString(ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH, ""));
+ File fsRecovery = new File(config.getString(ConfigConstants.ZOOKEEPER_RECOVERY_PATH, ""));
LOG.info("Checking " + fsRecovery);
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-tests/src/test/java/org/apache/flink/test/runtime/leaderelection/ZooKeeperLeaderElectionITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/leaderelection/ZooKeeperLeaderElectionITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/leaderelection/ZooKeeperLeaderElectionITCase.java
index 5840a98..10417c8 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/leaderelection/ZooKeeperLeaderElectionITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/leaderelection/ZooKeeperLeaderElectionITCase.java
@@ -95,7 +95,7 @@ public class ZooKeeperLeaderElectionITCase extends TestLogger {
configuration.setInteger(ConfigConstants.LOCAL_NUMBER_JOB_MANAGER, numJMs);
configuration.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, numTMs);
configuration.setString(ConfigConstants.STATE_BACKEND, "filesystem");
- configuration.setString(ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH, tempDirectory.getAbsoluteFile().toURI().toString());
+ configuration.setString(ConfigConstants.ZOOKEEPER_RECOVERY_PATH, tempDirectory.getAbsoluteFile().toURI().toString());
ForkableFlinkMiniCluster cluster = new ForkableFlinkMiniCluster(configuration);
@@ -144,7 +144,7 @@ public class ZooKeeperLeaderElectionITCase extends TestLogger {
configuration.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, numTMs);
configuration.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlotsPerTM);
configuration.setString(ConfigConstants.STATE_BACKEND, "filesystem");
- configuration.setString(ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH, tempDirectory.getAbsoluteFile().toURI().toString());
+ configuration.setString(ConfigConstants.ZOOKEEPER_RECOVERY_PATH, tempDirectory.getAbsoluteFile().toURI().toString());
// we "effectively" disable the automatic RecoverAllJobs message and sent it manually to make
// sure that all TMs have registered to the JM prior to issueing the RecoverAllJobs message
http://git-wip-us.apache.org/repos/asf/flink/blob/ab2895fa/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java b/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
index ffb43f8..f68b141 100644
--- a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
+++ b/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
@@ -112,11 +112,11 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
String fsStateHandlePath = tmp.getRoot().getPath();
flinkYarnClient.setFlinkConfigurationObject(GlobalConfiguration.getConfiguration());
- flinkYarnClient.setDynamicPropertiesEncoded("recovery.mode=zookeeper@@ha.zookeeper.quorum=" +
+ flinkYarnClient.setDynamicPropertiesEncoded("recovery.mode=zookeeper@@recovery.zookeeper.quorum=" +
zkServer.getConnectString() + "@@yarn.application-attempts=" + numberApplicationAttempts +
"@@" + ConfigConstants.STATE_BACKEND + "=FILESYSTEM" +
"@@" + FsStateBackendFactory.CHECKPOINT_DIRECTORY_URI_CONF_KEY + "=" + fsStateHandlePath + "/checkpoints" +
- "@@" + ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH + "=" + fsStateHandlePath + "/recovery");
+ "@@" + ConfigConstants.ZOOKEEPER_RECOVERY_PATH + "=" + fsStateHandlePath + "/recovery");
flinkYarnClient.setConfigurationFilePath(new Path(confDirPath + File.separator + "flink-conf.yaml"));
AbstractFlinkYarnCluster yarnCluster = null;