{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/46856258","id":46856258,"diff_hunk":"@@ -440,4 +478,228 @@ private void configureExecutionRetryDelay() {\n \t\tlong executionRetryDelay = streamGraph.getExecutionConfig().getExecutionRetryDelay();\n \t\tjobGraph.setExecutionRetryDelay(executionRetryDelay);\n \t}\n+\n+\t// ------------------------------------------------------------------------\n+\n+\t/**\n+\t * Returns a map with a hash for each {@link StreamNode} of the {@link\n+\t * StreamGraph}. The hash is used as the {@link JobVertexID} in order to\n+\t * identify nodes across job submissions if they didn't change.\n+\t *\n+\t * <p>The complete {@link StreamGraph} is traversed. The hash is either\n+\t * computed from the transformation's user-specified id (see\n+\t * {@link StreamTransformation#getUid()}) or generated in a deterministic way.\n+\t *\n+\t * <p>The generated hash is deterministic with respect to:\n+\t * <ul>\n+\t * <li>node-local properties (like parallelism, UDF, node ID),\n+\t * <li>chained output nodes, and\n+\t * <li>input nodes hashes\n+\t * </ul>\n+\t *\n+\t * @return A map from {@link StreamNode#id} to hash as 16-byte array.\n+\t */\n+\tprivate Map<Integer, byte[]> traverseStreamGraphAndGenerateHashes() {\n+\t\t// The hash function used to generate the hash\n+\t\tfinal HashFunction hashFunction = Hashing.murmur3_128(0);\n+\t\tfinal Map<Integer, byte[]> hashes = new HashMap<>();\n+\n+\t\tSet<Integer> visited = new HashSet<>();\n+\t\tQueue<StreamNode> remaining = new ArrayDeque<>();\n+\n+\t\t// We need to make the source order deterministic. This depends on the\n+\t\t// ordering of the sources in the Environment, e.g. if a source X is\n+\t\t// added before source Y, X will have a lower ID than Y (assigned by a\n+\t\t// static counter).","path":"flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java","position":262,"original_position":262,"commit_id":"52031da26d179bd4e5f85f2d0dae8b9f8cad5b02","original_commit_id":"25e6a084d3dd6eecad5cbc249c3a2a5f4b65ed8b","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":"I think this is not even necessary if we don't include the StreamNode ID in the hash for the sources.","created_at":"2015-12-07T18:25:44Z","updated_at":"2015-12-07T18:25:44Z","html_url":"https://github.com/apache/flink/pull/1434#discussion_r46856258","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/1434","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/46856258"},"html":{"href":"https://github.com/apache/flink/pull/1434#discussion_r46856258"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/1434"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/1434","id":52661336,"html_url":"https://github.com/apache/flink/pull/1434","diff_url":"https://github.com/apache/flink/pull/1434.diff","patch_url":"https://github.com/apache/flink/pull/1434.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/1434","number":1434,"state":"open","locked":false,"title":"[FLINK-2976] Allow to trigger checkpoints manually","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":"This PR contains **documentation**: https://github.com/uce/flink/blob/2976-savepoints/docs/apis/savepoints.md\r\n\r\n**In a nutshell**, savepoints `(*)` are **manually triggered checkpoints**, which take a snapshot of a program and write it out to an external state backend. This allows you to stop and resume your program without loosing intermediate state.\r\n\r\n**Why is this nice?** Because you don't have to replay everything when you redeploy your long running streaming job after changing it or updating to a newer Flink version.\r\n\r\n`(*)` Initially I wrote it as sa**F**epoints, but then settled on sa**V**epoints after stubmling across a related feature in [an Oracle SQL reference](https://docs.oracle.com/cd/B19306_01/appdev.102/b14261/savepoint_statement.htm). What do you think? :smile: http://doodle.com/poll/2z2cp9hxu7eucdsz\r\n\r\n## Example\r\n\r\nStart your stateful streaming program via `flink/bin run ...`.\r\n\r\n```\r\n$ bin/flink list\r\n------------------ Running/Restarting Jobs -------------------\r\n04.12.2015 13:51:10 : 46da86f25ca8daa1bbff8ccae64d53af : Flink Streaming Job (RUNNING)\r\n--------------------------------------------------------------\r\n```\r\n\r\nWait for some checkpoints to complete:\r\n\r\n```\r\n$ tail -f log/flink-hadoop-client-uce-m.log\r\n...\r\n13:50:59,806 INFO  org.apache.flink.runtime.jobmanager.JobManager - Status of job 46da86f25ca8daa1bbff8ccae64d53af (Flink Streaming Job) changed to RUNNING.\r\n...\r\n13:55:37,225 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Triggering checkpoint 1 @ 1449150937225\r\n13:55:37,581 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Completed checkpoint 1\r\n13:55:42,225 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Triggering checkpoint 2 @ 1449150942225\r\n13:55:42,328 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Completed checkpoint 2\r\n...\r\n13:56:27,225 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Triggering checkpoint 11 @ 1449150987225\r\n13:56:27,237 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Completed checkpoint 11\r\n...\r\n```\r\n\r\nTrigger a savepoint and cancel the job:\r\n\r\n```\r\n$ bin/flink savepoint 46da86f25ca8daa1bbff8ccae64d53af\r\nTriggering savepoint for job 46da86f25ca8daa1bbff8ccae64d53af. Waiting for response...\r\nSavepoint completed. Path: jobmanager://savepoints/1\r\nYou can resume your program from this savepoint with the run command.\r\n\r\n$ bin/flink cancel 46da86f25ca8daa1bbff8ccae64d53af\r\n```\r\n\r\nNow you can restart the program from the savepoint:\r\n\r\n```\r\n$ bin/flink run --fromSavepoint jobmanager://savepoints/1 ...\r\n```\r\n\r\nThis will resume the application from the state of the savepoint.\r\n\r\n## Changes to Flink\r\n\r\nI focussed on **not changing any major Flink component** for this. Savepoints use the same checkpointing mechanism as the periodic checkpoints with some plumbing around it.\r\n\r\n### Savepoint coordinator\r\n\r\nIn addition to the `CheckpointCoordinator`, we add another instance of the `CheckpointCoordinator` called `SavepointCoordinator`. This class extends the regular coordinator and registers some callbacks on shutdown, fully ack'ed checkpoint, and cancelled checkpoint. For this, I've added three callback methods to the checkpoint coordinator, which are overwritten by the savepoint coordinator. With two separate coordinators, periodic checkpoints and savepoints don't interfere with each other.\r\n\r\nThe savepoint coordinator manages a map of `checkpoint ID => future`. The futures are completed when the checkpoint is ack'ed or cancelled (or the coordinator shuts down altogether).\r\n\r\n#### Restoring\r\n\r\nRestore happens on job submission if a savepoint path is provided in the `JobSnapshottingSettings`. The restore mechanism is similar to the regular checkpoint restore, but with some further sanity checks to ensure that the state to task mapping is correct (see below). All state has to be mapped to the restored program.\r\n\r\n### JobManagerMessages\r\n\r\nAdded `TriggerSavepoint(JobID)` and `DisposeSavepoint(String)` Akka messages to the job manager. They trigger and dispose the savepoints respectively. These operations work asynchronously and respond the the request when the savepoint futures complete. The requests are triggered by the user (see CLI frontend).\r\n \r\n### Hashing of StreamNodes\r\n\r\nThe state to task mapping of checkpoints happens via `(jobVertexID, subtaskIndex)`. With this change, the jobVertexIDs of streaming programs are generated deterministically with respect to the structure of the program. This is needed to make sure that a restore with a new program can map the savepoint state to the tasks.\r\n\r\nThe hash starts from the sources and takes multiple things into account:\r\n- parallelism\r\n- user function class\r\n- hash of the input\r\n- hash of the outputs\r\n- stream node ID\r\n\r\nThe automatic generations makes sure that you can just use the savepoints, but it is actually *not recommended*, because you cannot change the program in any meaningful way (except changing the user function internals).\r\n\r\nThat's why the **recommended option** is to specify a unique ID as input to the hasher on the DataStream API level:\r\n\r\n```\r\nDataStream<String> stream = env.\r\n  // Stateful source (e.g. Kafka) with ID\r\n  .addSource(new StatefulSource()).uid(\"source-id\")\r\n  .shuffle()\r\n  // The stateful mapper with ID\r\n  .map(new StatefulMapper()).uid(\"mapper-id\")\r\n\r\n// Stateless sink (no specific ID required)\r\nstream.print()\r\n```\r\n\r\nIf you give IDs to all stateful operators, you can happily re-arrange and change the topology (except for parallelism, see below).\r\n\r\n### Application ID and DbStateBackend\r\n\r\nSavepoints are pairs of `(ApplicationID, CompletedCheckpoint)`. I've added a new `ApplicationID` to allow scoping tasks across multiple job submissions (which have changing job IDs). This is for example required to restore state in the `DbStateBackend`. After consulting with @gyfora I've changed all references to `JobID` in `DbStateBackend` to  `ApplicationID`.\r\n\r\nThe ApplicationID is assigned in the `ExecutionGraph` only and is reset to the application ID of the savepoint if there is a restore operation. This touches almost nothing of the existing code and it is only propagated to the `TaskDeploymentDescriptor` and `Environment` of the `Task` instances.\r\n\r\n### State storage\r\n\r\nThe state storage **does not** instantiate the regular state backend on the job manager. It is essentially a set of a few helper classes, which allow to put and get some state to the file system or the job manager's heap. I think this is fine for now, because I didn't want to make changes to the central state abstractions, which are kind of in flux right now. But we should think about it in the future.\r\n\r\n### Configuration and CLIFrontend\r\n\r\nThis works out of the box if the job is using checkpointing. The default state backend for savepoints is `jobmanager`, which allows to stop and restore a program while the same job manager is running.\r\n\r\nFor configuration, there are two new keys:\r\n```\r\nstate.backend.savepoints\r\nstate.backend.savepoints.fs.dir\r\n```\r\nIf you don't specify these, the regular state backend configuration is used with the `jobmanager` as a fallback if no viable config is found.\r\n","created_at":"2015-12-04T12:04:16Z","updated_at":"2015-12-07T18:25:44Z","closed_at":null,"merged_at":null,"merge_commit_sha":"1bf10e54b4b23cf532465c923b0e94e080540b13","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/1434/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/1434/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/1434/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/52031da26d179bd4e5f85f2d0dae8b9f8cad5b02","head":{"label":"uce:2976-savepoints","ref":"2976-savepoints","sha":"52031da26d179bd4e5f85f2d0dae8b9f8cad5b02","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-12-07T16:10:37Z","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":69431,"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":1,"forks":0,"open_issues":1,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"868f97cf8ae4e0f8cd9b3c8bbb43279ea908bf3e","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-12-07T16:04:37Z","pushed_at":"2015-12-07T18:00:31Z","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":86282,"stargazers_count":790,"watchers_count":790,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":526,"mirror_url":"git://git.apache.org/incubator-flink.git","open_issues_count":67,"forks":526,"open_issues":67,"watchers":790,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/1434"},"html":{"href":"https://github.com/apache/flink/pull/1434"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/1434"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/1434/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/1434/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/1434/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/52031da26d179bd4e5f85f2d0dae8b9f8cad5b02"}}}}