{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/41152813","id":41152813,"diff_hunk":"@@ -0,0 +1,422 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one\n+ * or more contributor license agreements.  See the NOTICE file\n+ * distributed with this work for additional information\n+ * regarding copyright ownership.  The ASF licenses this file\n+ * to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance\n+ * with the License.  You may obtain a copy of the License at\n+ *\n+ *     http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package org.apache.flink.runtime.zookeeper;\n+\n+import org.apache.curator.framework.CuratorFramework;\n+import org.apache.curator.framework.api.BackgroundCallback;\n+import org.apache.curator.utils.ZKPaths;\n+import org.apache.flink.runtime.state.StateHandle;\n+import org.apache.flink.runtime.state.StateHandleProvider;\n+import org.apache.flink.util.InstantiationUtil;\n+import org.apache.zookeeper.CreateMode;\n+import org.apache.zookeeper.KeeperException;\n+import org.apache.zookeeper.data.Stat;\n+\n+import java.io.Serializable;\n+import java.util.ArrayList;\n+import java.util.ConcurrentModificationException;\n+import java.util.List;\n+\n+import static com.google.common.base.Preconditions.checkNotNull;\n+\n+/**\n+ * State handles backed by ZooKeeper.\n+ *\n+ * <p>Added state is persisted via {@link StateHandle}s, which in turn are written to\n+ * ZooKeeper. This level of indirection is necessary to keep the amount of data in ZooKeeper\n+ * small. ZooKeeper is build for data in the KB range whereas state can grow to multiple MBs.\n+ *\n+ * <p>State modifications require some care, because it is possible that certain failures bring\n+ * the state handle backend and ZooKeeper out of sync.\n+ *\n+ * <p>ZooKeeper holds the ground truth about state handles, i.e. the following holds:\n+ *\n+ * <pre>\n+ * State handle in ZooKeeper => State handle exists\n+ * </pre>\n+ *\n+ * But not:\n+ *\n+ * <pre>\n+ * State handle exists => State handle in ZooKeeper\n+ * </pre>\n+ *\n+ * There can be lingering state handles when failures happen during operation. They\n+ * need to be cleaned up manually (see <a href=\"https://issues.apache.org/jira/browse/FLINK-2513\">\n+ * FLINK-2513</a> about a possible way to overcome this).\n+ *\n+ * @param <T> Type of state\n+ */\n+public class ZooKeeperStateHandleStore<T extends Serializable> {\n+\n+\t/** Curator ZooKeeper client */\n+\tprivate final CuratorFramework client;\n+\n+\t/** State handle provider */\n+\tprivate final StateHandleProvider<T> stateHandleProvider;\n+\n+\t/**\n+\t * Creates a {@link ZooKeeperStateHandleStore}.\n+\t *\n+\t * @param client              The Curator ZooKeeper client. <strong>Important:</strong> It is\n+\t *                            expected that the client's namespace ensures that the root\n+\t *                            path is exclusive for all state handles managed by this\n+\t *                            instance, e.g. <code>client.usingNamespace(\"/stateHandles\")</code>\n+\t * @param stateHandleProvider The state handle provider for the state\n+\t */\n+\tpublic ZooKeeperStateHandleStore(\n+\t\t\tCuratorFramework client,\n+\t\t\tStateHandleProvider<T> stateHandleProvider) {\n+\n+\t\tthis.client = checkNotNull(client, \"Curator client\");\n+\t\tthis.stateHandleProvider = checkNotNull(stateHandleProvider, \"State handle provider\");\n+\t}\n+\n+\t/**\n+\t * Creates a state handle and stores it in ZooKeeper with create mode {@link\n+\t * CreateMode#PERSISTENT}.\n+\t *\n+\t * @see #add(String, Serializable, CreateMode)\n+\t */\n+\tpublic ZooKeeperStateHandle<T> add(String pathInZooKeeper, T state) throws Exception {\n+\t\treturn add(pathInZooKeeper, state, CreateMode.PERSISTENT);\n+\t}\n+\n+\t/**\n+\t * Creates a state handle and stores it in ZooKeeper.\n+\t *\n+\t * <p><strong>Important</strong>: This will <em>not</em> store the actual state in\n+\t * ZooKeeper, but create a state handle and store it in ZooKeeper. This level of indirection\n+\t * makes sure that data in ZooKeeper is small.\n+\t *\n+\t * @param pathInZooKeeper Destination path in ZooKeeper (expected to *not* exist yet and\n+\t *                        start with a '/')\n+\t * @param state           State to be added\n+\t * @param createMode      The create mode for the new path in ZooKeeper\n+\t * @return Created {@link ZooKeeperStateHandle}\n+\t * @throws Exception\n+\t */\n+\tpublic ZooKeeperStateHandle<T> add(String pathInZooKeeper, T state, CreateMode createMode) throws Exception {\n+\t\tcheckNotNull(pathInZooKeeper, \"Path in ZooKeeper\");\n+\t\tcheckNotNull(state, \"State\");\n+\n+\t\t// Create the state handle. Nothing persisted yet.\n+\t\tStateHandle<T> stateHandle = stateHandleProvider.createStateHandle(state);\n+\n+\t\tboolean success = false;\n+\n+\t\ttry {\n+\t\t\t// Serialize the state handle. This writes the state to the backend.\n+\t\t\tbyte[] serializedStateHandle = InstantiationUtil.serializeObject(stateHandle);\n+\n+\t\t\t// Write state handle (not the actual state) to ZooKeeper. This is expected to be\n+\t\t\t// smaller than the state itself. This level of indirection makes sure that data in\n+\t\t\t// ZooKeeper is small, because ZooKeeper is designed for data in the KB range, but\n+\t\t\t// the state can be larger.\n+\t\t\tclient\n+\t\t\t\t\t.create()\n+\t\t\t\t\t.withMode(createMode)\n+\t\t\t\t\t.forPath(pathInZooKeeper, serializedStateHandle);\n+\n+\t\t\tsuccess = true;\n+\n+\t\t\treturn new ZooKeeperStateHandle<>(stateHandle, pathInZooKeeper);\n+\t\t}\n+\t\tfinally {\n+\t\t\tif (!success) {\n+\t\t\t\t// Cleanup the state handle if it was not written to ZooKeeper.\n+\t\t\t\tif (stateHandle != null) {\n+\t\t\t\t\tstateHandle.discardState();\n+\t\t\t\t}\n+\t\t\t}\n+\t\t}\n+\t}\n+\n+\t/**\n+\t * Replaces a state handle in ZooKeeper and discards the old state handle.\n+\t *\n+\t * <p><strong>Important</strong>: This method will only discard the state handle and not the\n+\t * state itself. Don't forget to run custom cleanup code of the state, if necessary.\n+\t *\n+\t * <pre>\n+\t * T state = get(path).getState();\n+\t * state.discard(); // Custom clean up\n+\t * replace(path, version, newState)\n+\t * </pre>\n+\t *\n+\t * @param pathInZooKeeper Destination path in ZooKeeper (expected to exist and start with a '/')\n+\t * @param expectedVersion Expected version of the node to replace\n+\t * @param state           The new state to replace the old one\n+\t */\n+\tpublic void replace(String pathInZooKeeper, int expectedVersion, T state) throws Exception {\n+\t\tcheckNotNull(pathInZooKeeper, \"Path in ZooKeeper\");\n+\t\tcheckNotNull(state, \"State\");\n+\n+\t\tStateHandle<T> oldStateHandle = get(pathInZooKeeper);\n+\n+\t\tStateHandle<T> stateHandle = stateHandleProvider.createStateHandle(state);\n+\n+\t\tboolean success = false;\n+\n+\t\ttry {\n+\t\t\t// Serialize the new state handle. This writes the state to the backend.\n+\t\t\tbyte[] serializedStateHandle = InstantiationUtil.serializeObject(stateHandle);\n+\n+\t\t\t// Replace state handle in ZooKeeper.\n+\t\t\tclient.setData()\n+\t\t\t\t\t.withVersion(expectedVersion)\n+\t\t\t\t\t.forPath(pathInZooKeeper, serializedStateHandle);\n+\n+\t\t\tsuccess = true;\n+\t\t}\n+\t\tfinally {\n+\t\t\tif (success) {\n+\t\t\t\toldStateHandle.discardState();\n+\t\t\t}\n+\t\t\telse {\n+\t\t\t\tstateHandle.discardState();\n+\t\t\t}\n+\t\t}\n+\t}\n+\n+\t/**\n+\t * Returns the version of the node if it exists or <code>-1</code> if it doesn't.\n+\t *\n+\t * @param pathInZooKeeper Path in ZooKeeper to check\n+\t * @return Version of the ZNode if the path exists, <code>-1</code> otherwise.\n+\t * @throws Exception\n+\t */\n+\tpublic int exists(String pathInZooKeeper) throws Exception {\n+\t\tcheckNotNull(pathInZooKeeper, \"Path in ZooKeeper\");\n+\n+\t\tStat stat = client.checkExists().forPath(pathInZooKeeper);\n+\n+\t\tif (stat != null) {\n+\t\t\treturn stat.getVersion();\n+\t\t}\n+\n+\t\treturn -1;\n+\t}\n+\n+\t/**\n+\t * Gets a state handle from ZooKeeper.\n+\t *\n+\t * @param pathInZooKeeper Path in ZooKeeper to get the state handle from (expected to\n+\t *                        exist and start with a '/').\n+\t * @return The state handle\n+\t * @throws Exception\n+\t */\n+\t@SuppressWarnings(\"unchecked\")\n+\tpublic StateHandle<T> get(String pathInZooKeeper) throws Exception {\n+\t\tcheckNotNull(pathInZooKeeper, \"Path in ZooKeeper\");\n+\n+\t\tbyte[] data = client.getData().forPath(pathInZooKeeper);\n+\n+\t\treturn (StateHandle<T>) InstantiationUtil.deserializeObject(data, ClassLoader\n+\t\t\t\t.getSystemClassLoader());\n+\t}\n+\n+\t/**\n+\t * Gets all available state handles from ZooKeeper.\n+\t *\n+\t * @return All state handles from ZooKeeper.\n+\t * @throws ConcurrentModificationException\n+\t */\n+\t@SuppressWarnings(\"unchecked\")\n+\tpublic List<ZooKeeperStateHandle<T>> getAll() throws Exception {\n+\t\t// Initial cVersion (number of changes to the children of this node)\n+\t\tint initialCVersion = client.checkExists().forPath(\"/\").getCversion();\n+\n+\t\tList<String> children = client.getChildren().forPath(\"/\");\n+\n+\t\tfinal List<ZooKeeperStateHandle<T>> stateHandles = new ArrayList<>(children.size());\n+\n+\t\tfor (String path : children) {\n+\t\t\tpath = \"/\" + path;\n+\n+\t\t\ttry {\n+\t\t\t\tfinal StateHandle<T> stateHandle = get(path);\n+\t\t\t\tstateHandles.add(new ZooKeeperStateHandle(stateHandle, path));\n+\t\t\t}\n+\t\t\tcatch (KeeperException.NoNodeException e) {\n+\t\t\t\tthrow new ConcurrentModificationException(\"Modification during getAll()\", e);\n+\t\t\t}\n+\t\t}\n+\n+\t\tverifyExpectedCVersion(initialCVersion, \"/\", \"Modification during getAll()\");\n+\n+\t\treturn stateHandles;\n+\t}\n+\n+\t/**\n+\t * Gets all available state handles from ZooKeeper sorted by name (ascending).\n+\t *\n+\t * @return All state handles in ZooKeeper.\n+\t */\n+\t@SuppressWarnings(\"unchecked\")\n+\tpublic List<ZooKeeperStateHandle<T>> getAllSortedByName() throws Exception {\n+\t\t// Initial cVersion (number of changes to the children of this node)\n+\t\tint initialCVersion = client.checkExists().forPath(\"/\").getCversion();\n+\n+\t\tList<String> children = ZKPaths.getSortedChildren(\n+\t\t\t\tclient.getZookeeperClient().getZooKeeper(),\n+\t\t\t\tZKPaths.fixForNamespace(client.getNamespace(), \"/\"));\n+\n+\t\tfinal List<ZooKeeperStateHandle<T>> stateHandles = new ArrayList<>(children.size());\n+\n+\t\tfor (String path : children) {\n+\t\t\tpath = \"/\" + path;\n+\t\t\ttry {\n+\t\t\t\tfinal StateHandle<T> stateHandle = get(path);\n+\t\t\t\tstateHandles.add(new ZooKeeperStateHandle(stateHandle, path));\n+\t\t\t}\n+\t\t\tcatch (KeeperException.NoNodeException e) {\n+\t\t\t\tthrow new ConcurrentModificationException(\"Modification during getAllSortedByName()\", e);\n+\t\t\t}\n+\t\t}\n+\n+\t\tverifyExpectedCVersion(initialCVersion, \"/\", \"Modification during getAllSortedByName()\");\n+\n+\t\treturn stateHandles;\n+\t}\n+\n+\t/**\n+\t * Removes a state handle from ZooKeeper.\n+\t *\n+\t * <p><stong>Important</stong>: this does not discard the state handle. If you want to\n+\t * discard the state handle call {@link #discard(String)}.\n+\t *\n+\t * @param pathInZooKeeper Path of state handle to remove (expected to start with a '/')\n+\t * @throws Exception\n+\t */\n+\tpublic void remove(String pathInZooKeeper) throws Exception {\n+\t\tcheckNotNull(pathInZooKeeper, \"Path in ZooKeeper\");\n+\n+\t\tclient\n+\t\t\t\t.delete()\n+\t\t\t\t.deletingChildrenIfNeeded()\n+\t\t\t\t.forPath(pathInZooKeeper);\n+\t}\n+\n+\t/**\n+\t * Removes a state handle from ZooKeeper asynchronously.\n+\t *\n+\t * <p><stong>Important</stong>: this does not discard the state handle. If you want to\n+\t * discard the state handle call {@link #discard(String)}.\n+\t *\n+\t * @param pathInZooKeeper Path of state handle to remove (expected to start with a '/')\n+\t * @param callback        The callback after the operation finishes\n+\t * @throws Exception\n+\t */\n+\tpublic void remove(String pathInZooKeeper, BackgroundCallback callback) throws Exception {\n+\t\tcheckNotNull(pathInZooKeeper, \"Path in ZooKeeper\");\n+\t\tcheckNotNull(callback, \"Background callback\");\n+\n+\t\tclient\n+\t\t\t\t.delete()\n+\t\t\t\t.deletingChildrenIfNeeded()\n+\t\t\t\t.inBackground(callback)\n+\t\t\t\t.forPath(pathInZooKeeper);\n+\t}\n+\n+\t/**\n+\t * Discards a state handle and removes it from ZooKeeper.\n+\t *\n+\t * <p><strong>Important</strong>: This method will only discard the state handle and not the\n+\t * state itself. Don't forget to run custom cleanup code of the state, if necessary.\n+\t *\n+\t * <pre>\n+\t * T state = get(path).getState();\n+\t * state.discard(); // Custom clean up\n+\t * discard(path)\n+\t * </pre>\n+\t *\n+\t * <p>If you only want to remove the state handle in ZooKeeper call {@link #remove(String)}.\n+\t *\n+\t * @param pathInZooKeeper Path of state handle to discard (expected to start with a '/')\n+\t * @throws Exception\n+\t */\n+\tpublic void discard(String pathInZooKeeper) throws Exception {\n+\t\tcheckNotNull(pathInZooKeeper, \"Path in ZooKeeper\");\n+\n+\t\tStateHandle<T> stateHandle = get(pathInZooKeeper);\n+\n+\t\t// Delete the state handle from ZooKeeper first\n+\t\tclient\n+\t\t\t\t.delete()\n+\t\t\t\t.deletingChildrenIfNeeded()\n+\t\t\t\t.forPath(pathInZooKeeper);\n+\n+\t\t// Discard the state handle only after it has been successfully deleted from ZooKeeper.\n+\t\t// Otherwise we might enter an illegal state after failures (with a state handle in\n+\t\t// ZooKeeper, which has already been discarded).\n+\t\tstateHandle.discardState();\n+\t}\n+\n+\t/**\n+\t * Discards all available state handles and removes them from ZooKeeper.\n+\t *\n+\t * <p><strong>Important</strong>: This method will only discard the state handles and not the\n+\t * states itself. Don't forget to run custom cleanup code of the states, if necessary.\n+\t *\n+\t * <pre>\n+\t * for (StateHandle<T> handle : getAll()) {\n+\t *     handle.getState().discard(); // Custom clean up\n+\t * }\n+\t * discardAll()\n+\t * </pre>\n+\t *\n+\t * @throws Exception\n+\t */\n+\tpublic void discardAll() throws Exception {\n+\t\tfinal List<ZooKeeperStateHandle<T>> allStateHandles = getAll();\n+\n+\t\tZKPaths.deleteChildren(\n+\t\t\t\tclient.getZookeeperClient().getZooKeeper(),\n+\t\t\t\tZKPaths.fixForNamespace(client.getNamespace(), \"/\"),\n+\t\t\t\tfalse);\n+\n+\t\t// Discard the state handles only after they have been successfully deleted from ZooKeeper.\n+\t\tfor (ZooKeeperStateHandle<T> stateHandle : allStateHandles) {\n+\t\t\tstateHandle.getStateHandle().discardState();\n+\t\t}\n+\t}\n+\n+\t/**\n+\t * Throws a {@link ConcurrentModificationException} if the cVersion of the specified path\n+\t * does not match the expected cVersion.\n+\t *\n+\t * @param expectedCVersion The expected cVersion\n+\t * @param pathInZooKeeper  Path in ZooKeeper to check (expected to exist and start with '/')\n+\t * @param errorMessage     The error message of the thrown Exception\n+\t * @throws Exception\n+\t */\n+\tprivate void verifyExpectedCVersion(\n+\t\t\tint expectedCVersion,\n+\t\t\tString pathInZooKeeper,\n+\t\t\tString errorMessage) throws Exception {","path":"flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java","position":413,"original_position":413,"commit_id":"a50899b5fea4289c51ff09395b6300e5ff410f05","original_commit_id":"a50899b5fea4289c51ff09395b6300e5ff410f05","user":{"login":"uce","id":1756620,"avatar_url":"https://avatars.githubusercontent.com/u/1756620?v=3","gravatar_id":"","url":"https://api.github.com/users/uce","html_url":"https://github.com/uce","followers_url":"https://api.github.com/users/uce/followers","following_url":"https://api.github.com/users/uce/following{/other_user}","gists_url":"https://api.github.com/users/uce/gists{/gist_id}","starred_url":"https://api.github.com/users/uce/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/uce/subscriptions","organizations_url":"https://api.github.com/users/uce/orgs","repos_url":"https://api.github.com/users/uce/repos","events_url":"https://api.github.com/users/uce/events{/privacy}","received_events_url":"https://api.github.com/users/uce/received_events","type":"User","site_admin":false},"body":"Most of the generic Exception comes from Curator. I have to check whether it is feasible to provide more specific ones.","created_at":"2015-10-05T14:53:02Z","updated_at":"2015-10-05T14:53:02Z","html_url":"https://github.com/apache/flink/pull/1153#discussion_r41152813","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/1153","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/41152813"},"html":{"href":"https://github.com/apache/flink/pull/1153#discussion_r41152813"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/1153"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/1153","id":45519067,"html_url":"https://github.com/apache/flink/pull/1153","diff_url":"https://github.com/apache/flink/pull/1153.diff","patch_url":"https://github.com/apache/flink/pull/1153.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/1153","number":1153,"state":"open","locked":false,"title":"[FLINK-2354] Add job graph and checkpoint recovery","user":{"login":"uce","id":1756620,"avatar_url":"https://avatars.githubusercontent.com/u/1756620?v=3","gravatar_id":"","url":"https://api.github.com/users/uce","html_url":"https://github.com/uce","followers_url":"https://api.github.com/users/uce/followers","following_url":"https://api.github.com/users/uce/following{/other_user}","gists_url":"https://api.github.com/users/uce/gists{/gist_id}","starred_url":"https://api.github.com/users/uce/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/uce/subscriptions","organizations_url":"https://api.github.com/users/uce/orgs","repos_url":"https://api.github.com/users/uce/repos","events_url":"https://api.github.com/users/uce/events{/privacy}","received_events_url":"https://api.github.com/users/uce/received_events","type":"User","site_admin":false},"body":"## tl;dr\r\n\r\nThis PR introduces `JobGraph` and `SuccessfulCheckpoint` recovery for submitted programs in case of JobManager failures.\r\n\r\n## General Idea\r\n\r\nThe general idea is to persist job graphs and successful checkpoints in ZooKeeper.\r\n\r\nWe have introduced JobManager high availability via ZooKeeper in #1016. My PR builds on top of this and adds initial support for program recovery. We can recover both programs and successful checkpoints in case of a JobManager failure as soon as a standby job manager is granted leadership.\r\n\r\nZooKeeper's sweet spot is rather small data (in KB range), but job graph and checkpoint state can grow larger. Therefore we don't directly persist the actual metadata, but use the state backend as a layer of indirection. We create state handles for the job graph and completed checkpoints and persist those. The state handle acts as a pointer to the actual data.\r\n\r\nAt the moment, only the file system state backend is supported for this. The state handles need to be accessible from both task and job managers (e.g. a DFS).\r\n\r\n## Configuration\r\n\r\nThe minimal required configuration:\r\n\r\n```bash\r\nrecovery.mode: ZOOKEEPER\r\nha.zookeeper.quorum: <ZooKeeper quroum peers>\r\nstate.backend: FILESYSTEM\r\nstate.backend.fs.dir.recovery: /path/to/recovery\r\n```\r\n\r\nI don't like the current configuration keys. Until the next release, I would like a more consistent naming, e.g. prefix everything with `recovery.zookeeper`.\r\n\r\n## ZooKeeper Nodes Overview\r\n\r\nOverview of ZNodes and components managing them:\r\n\r\n```bash\r\nO- /flink\r\n|\r\n+----O /flink/jobgraphs (SubmittedJobGraphs)\r\n|    |\r\n|    +----O /flink/jobgraphs/<job-id>\r\n|\r\n+----O /flink/checkpoints  (CompletedCheckpoints)\r\n|    |\r\n|    +----O /flink/checkpoints/<job-id>\r\n|    .    |\r\n|    .    +----O /flink/checkpoints/<job-id>/1\r\n|    .    |\r\n|    .    +----O /flink/checkpoints/<job-id>/N\r\n|\r\n+----O /flink/checkpoint-counter (CheckpointIDCounter)\r\n     |\r\n     +----O /flink/checkpoints/<job-id>\r\n```\r\n\r\n## Implementation\r\n\r\n### Submission vs. Recovery (JobManager and SubmittedJobGraphs)\r\n\r\n- `ZooKeeperSubmittedJobGraphs` manages `SubmittedJobGraph` state handles in ZooKeeper\r\n- Submission and recovery follow mostly the same code paths (see `JobManager#submitJob()`).\r\n- On (initial) submission:\r\n  - After writing to ZooKeeper the JM checks synchronously whether she is still leader.\r\n  - If not, the job is not scheduled for execution, but kept in ZooKeeper. Future leading JobManagers need to recover it. The client currently sees this as successful submission. The job is not removed in this case, because another job manager might recover between the write and remove. In such a case, a job would be running without being in ZooKeeper and without being acked to the client.\r\n- On recovery:\r\n  - Recovery is triggered on granted leadership via the configured execution delay between retries.\r\n  - All available jobs are scheduled for execution.\r\n- The ZNode for job graphs is monitored for modifications during operations. This way, a job manager can (eventually) detect if another job manager adds/removes a job and react to it.\r\n\r\n### CompletedCheckpoints\r\n\r\n- `ZooKeeperCompletedCheckpoints` manages `SuccessfulCheckpoint` state handles in ZooKeeper (per job). Note that a `SuccessfulCheckpoint` has pointers to further state handles in most cases. In this case, we add another layer of indirection.\r\n- Every completed checkpoint is added to ZooKeeper and identified by its checkpoint ID.\r\n- On recovery, the latest checkpoint is recovered. If more than one checkpoint is available, we still only recover one in order to make sure that the history of checkpoints is consistent (currently we retain only 1 checkpoint anyways, but if we ever chose to retain more) in corner cases, where multiple job managers run the same job with checkpointing for some time.\r\n\r\n### CheckpointIDCounter\r\n\r\n- `ZooKeeperCheckpointIDCounter` manages a shared counter in ZooKeeper (per job).\r\n- The `Checkpointed` interface requires ascending checkpoint IDs for each checkpoint.\r\n- We use a shared counter (per job) via a Curator recipe for this.\r\n\r\n### Akka messages\r\n\r\n- This PR introduces two new JobManager message types:\r\n  - RecoverAllJobs\r\n  - RecoverJob(JobID)\r\n- The ZooKeeper operations are blocking and all JobManager actor calls needs to make sure to *not* block the JobManager. I've tried to cover all cases, where a ZooKeeper operation is triggered.\r\n- For tests, I didn't manage to stop the JobManager actor w/o running the `postStop` method. Because this method has some cleanup logic (removing job graphs and checkpoints), all JobManager recovery tests run the JobManager as a separate `JobManagerProcess`. This is quite heavy weight. If someone knows a way to stop the actor w/o the `postStop` being called, it would be great to refactor this.\r\n\r\n## Next Steps\r\n\r\n- Behaviour on recovery via fixed delay is too simplistic.\r\n- Client is not fully integrated and submits jobs in detached mode if recovery mode is set to ZooKeeper.\r\n\r\n## Tests\r\n\r\n~~There was a Travis/AWS outage yesterday and I couldn't run as many builds as we should yet. I would like to run a couple of runs before we merge this.~~\r\n","created_at":"2015-09-21T07:08:11Z","updated_at":"2015-10-05T14:53:02Z","closed_at":null,"merged_at":null,"merge_commit_sha":"de74d27b75444116caebdefebfb564857199b396","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/1153/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/1153/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/1153/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/a50899b5fea4289c51ff09395b6300e5ff410f05","head":{"label":"uce:recovery-2354","ref":"recovery-2354","sha":"a50899b5fea4289c51ff09395b6300e5ff410f05","user":{"login":"uce","id":1756620,"avatar_url":"https://avatars.githubusercontent.com/u/1756620?v=3","gravatar_id":"","url":"https://api.github.com/users/uce","html_url":"https://github.com/uce","followers_url":"https://api.github.com/users/uce/followers","following_url":"https://api.github.com/users/uce/following{/other_user}","gists_url":"https://api.github.com/users/uce/gists{/gist_id}","starred_url":"https://api.github.com/users/uce/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/uce/subscriptions","organizations_url":"https://api.github.com/users/uce/orgs","repos_url":"https://api.github.com/users/uce/repos","events_url":"https://api.github.com/users/uce/events{/privacy}","received_events_url":"https://api.github.com/users/uce/received_events","type":"User","site_admin":false},"repo":{"id":20699435,"name":"flink","full_name":"uce/flink","owner":{"login":"uce","id":1756620,"avatar_url":"https://avatars.githubusercontent.com/u/1756620?v=3","gravatar_id":"","url":"https://api.github.com/users/uce","html_url":"https://github.com/uce","followers_url":"https://api.github.com/users/uce/followers","following_url":"https://api.github.com/users/uce/following{/other_user}","gists_url":"https://api.github.com/users/uce/gists{/gist_id}","starred_url":"https://api.github.com/users/uce/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/uce/subscriptions","organizations_url":"https://api.github.com/users/uce/orgs","repos_url":"https://api.github.com/users/uce/repos","events_url":"https://api.github.com/users/uce/events{/privacy}","received_events_url":"https://api.github.com/users/uce/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/uce/flink","description":"Mirror of Apache Flink","fork":true,"url":"https://api.github.com/repos/uce/flink","forks_url":"https://api.github.com/repos/uce/flink/forks","keys_url":"https://api.github.com/repos/uce/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/uce/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/uce/flink/teams","hooks_url":"https://api.github.com/repos/uce/flink/hooks","issue_events_url":"https://api.github.com/repos/uce/flink/issues/events{/number}","events_url":"https://api.github.com/repos/uce/flink/events","assignees_url":"https://api.github.com/repos/uce/flink/assignees{/user}","branches_url":"https://api.github.com/repos/uce/flink/branches{/branch}","tags_url":"https://api.github.com/repos/uce/flink/tags","blobs_url":"https://api.github.com/repos/uce/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/uce/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/uce/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/uce/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/uce/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/uce/flink/languages","stargazers_url":"https://api.github.com/repos/uce/flink/stargazers","contributors_url":"https://api.github.com/repos/uce/flink/contributors","subscribers_url":"https://api.github.com/repos/uce/flink/subscribers","subscription_url":"https://api.github.com/repos/uce/flink/subscription","commits_url":"https://api.github.com/repos/uce/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/uce/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/uce/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/uce/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/uce/flink/contents/{+path}","compare_url":"https://api.github.com/repos/uce/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/uce/flink/merges","archive_url":"https://api.github.com/repos/uce/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/uce/flink/downloads","issues_url":"https://api.github.com/repos/uce/flink/issues{/number}","pulls_url":"https://api.github.com/repos/uce/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/uce/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/uce/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/uce/flink/labels{/name}","releases_url":"https://api.github.com/repos/uce/flink/releases{/id}","created_at":"2014-06-10T19:56:02Z","updated_at":"2015-06-18T14:47:26Z","pushed_at":"2015-09-30T14:56:39Z","git_url":"git://github.com/uce/flink.git","ssh_url":"git@github.com:uce/flink.git","clone_url":"https://github.com/uce/flink.git","svn_url":"https://github.com/uce/flink","homepage":null,"size":63344,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"622c1be8c61c3a9a37c30454c717c99982edd236","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":20587599,"name":"flink","full_name":"apache/flink","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/flink","description":"Mirror of Apache Flink","fork":false,"url":"https://api.github.com/repos/apache/flink","forks_url":"https://api.github.com/repos/apache/flink/forks","keys_url":"https://api.github.com/repos/apache/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/flink/teams","hooks_url":"https://api.github.com/repos/apache/flink/hooks","issue_events_url":"https://api.github.com/repos/apache/flink/issues/events{/number}","events_url":"https://api.github.com/repos/apache/flink/events","assignees_url":"https://api.github.com/repos/apache/flink/assignees{/user}","branches_url":"https://api.github.com/repos/apache/flink/branches{/branch}","tags_url":"https://api.github.com/repos/apache/flink/tags","blobs_url":"https://api.github.com/repos/apache/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/flink/languages","stargazers_url":"https://api.github.com/repos/apache/flink/stargazers","contributors_url":"https://api.github.com/repos/apache/flink/contributors","subscribers_url":"https://api.github.com/repos/apache/flink/subscribers","subscription_url":"https://api.github.com/repos/apache/flink/subscription","commits_url":"https://api.github.com/repos/apache/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/flink/contents/{+path}","compare_url":"https://api.github.com/repos/apache/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/flink/merges","archive_url":"https://api.github.com/repos/apache/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/flink/downloads","issues_url":"https://api.github.com/repos/apache/flink/issues{/number}","pulls_url":"https://api.github.com/repos/apache/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/flink/labels{/name}","releases_url":"https://api.github.com/repos/apache/flink/releases{/id}","created_at":"2014-06-07T07:00:10Z","updated_at":"2015-10-05T04:19:42Z","pushed_at":"2015-10-05T14:42:49Z","git_url":"git://github.com/apache/flink.git","ssh_url":"git@github.com:apache/flink.git","clone_url":"https://github.com/apache/flink.git","svn_url":"https://github.com/apache/flink","homepage":null,"size":2527748,"stargazers_count":656,"watchers_count":656,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":446,"mirror_url":"git://git.apache.org/incubator-flink.git","open_issues_count":79,"forks":446,"open_issues":79,"watchers":656,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/1153"},"html":{"href":"https://github.com/apache/flink/pull/1153"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/1153"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/1153/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/1153/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/1153/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/a50899b5fea4289c51ff09395b6300e5ff410f05"}}}}