{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/41241923","id":41241923,"diff_hunk":"@@ -0,0 +1,159 @@\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.blob;\n+\n+import org.apache.commons.io.FileUtils;\n+import org.apache.flink.api.common.JobID;\n+import org.apache.flink.configuration.ConfigConstants;\n+import org.apache.flink.configuration.Configuration;\n+import org.apache.flink.runtime.jobmanager.RecoveryMode;\n+import org.junit.After;\n+import org.junit.Before;\n+import org.junit.Test;\n+\n+import java.io.File;\n+import java.io.InputStream;\n+import java.net.InetSocketAddress;\n+import java.util.Arrays;\n+import java.util.Random;\n+\n+import static org.junit.Assert.assertEquals;\n+\n+public class BlobRecoveryITCase {\n+\n+\tprivate File recoveryDir;\n+\n+\t@Before\n+\tpublic void setUp() throws Exception {\n+\t\trecoveryDir = new File(FileUtils.getTempDirectory(), \"BlobRecoveryITCaseDir\");\n+\t\tif (!recoveryDir.exists() && !recoveryDir.mkdirs()) {\n+\t\t\tthrow new IllegalStateException(\"Failed to create temp directory for test\");\n+\t\t}\n+\t}\n+\n+\t@After\n+\tpublic void cleanUp() throws Exception {\n+\t\tif (recoveryDir != null) {\n+\t\t\tFileUtils.deleteDirectory(recoveryDir);\n+\t\t}\n+\t}\n+\n+\t/**\n+\t * Tests that with {@link RecoveryMode#ZOOKEEPER} distributed JARs are recoverable from any\n+\t * participating BlobServer.\n+\t */\n+\t@Test\n+\tpublic void testBlobServerRecovery() throws Exception {\n+\t\tRandom rand = new Random();\n+\n+\t\tBlobServer[] server = new BlobServer[2];\n+\t\tInetSocketAddress[] serverAddress = new InetSocketAddress[2];\n+\t\tBlobClient client = null;\n+\n+\t\ttry {\n+\t\t\tConfiguration config = new Configuration();\n+\t\t\tconfig.setString(ConfigConstants.RECOVERY_MODE, \"ZOOKEEPER\");\n+\t\t\tconfig.setString(ConfigConstants.STATE_BACKEND, \"FILESYSTEM\");\n+\t\t\tconfig.setString(ConfigConstants.STATE_BACKEND_FS_RECOVERY_PATH, recoveryDir.getPath());\n+\n+\t\t\tfor (int i = 0; i < server.length; i++) {\n+\t\t\t\tserver[i] = new BlobServer(config);\n+\t\t\t\tserverAddress[i] = new InetSocketAddress(\"localhost\", server[i].getPort());\n+\t\t\t}\n+\n+\t\t\tclient = new BlobClient(serverAddress[0]);\n+\n+\t\t\t// Random data\n+\t\t\tbyte[] actual = new byte[1024];\n+\t\t\trand.nextBytes(actual);\n+\n+\t\t\tBlobKey[] keys = new BlobKey[2];\n+\n+\t\t\t// Put data\n+\t\t\tkeys[0] = client.put(actual); // Request 1\n+\t\t\tkeys[1] = client.put(actual, 32, 256); // Request 2\n+\n+\t\t\tJobID[] jobId = new JobID[] { new JobID(), new JobID() };\n+\t\t\tString[] testKey = new String[] { \"test-key-1\", \"test-key-2\" };\n+\n+\t\t\tclient.put(jobId[0], testKey[0], actual); // Request 3\n+\t\t\tclient.put(jobId[1], testKey[1], actual, 32, 256); // Request 4\n+\n+\t\t\t// Close the client and connect to the other server\n+\t\t\tclient.close();\n+\t\t\tclient = new BlobClient(serverAddress[1]);\n+\n+\t\t\t// Verify request 1\n+\t\t\ttry (InputStream is = client.get(keys[0])) {\n+\t\t\t\tbyte[] expected = new byte[actual.length];","path":"flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobRecoveryITCase.java","position":104,"original_position":104,"commit_id":"00ea6b02f15f00486c9541af08b8197c41dd94f7","original_commit_id":"00ea6b02f15f00486c9541af08b8197c41dd94f7","user":{"login":"tillrohrmann","id":5756858,"avatar_url":"https://avatars.githubusercontent.com/u/5756858?v=3","gravatar_id":"","url":"https://api.github.com/users/tillrohrmann","html_url":"https://github.com/tillrohrmann","followers_url":"https://api.github.com/users/tillrohrmann/followers","following_url":"https://api.github.com/users/tillrohrmann/following{/other_user}","gists_url":"https://api.github.com/users/tillrohrmann/gists{/gist_id}","starred_url":"https://api.github.com/users/tillrohrmann/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tillrohrmann/subscriptions","organizations_url":"https://api.github.com/users/tillrohrmann/orgs","repos_url":"https://api.github.com/users/tillrohrmann/repos","events_url":"https://api.github.com/users/tillrohrmann/events{/privacy}","received_events_url":"https://api.github.com/users/tillrohrmann/received_events","type":"User","site_admin":false},"body":"`expected` should be the ground truth and `actual` the value you read.","created_at":"2015-10-06T09:16:11Z","updated_at":"2015-10-06T09:16:11Z","html_url":"https://github.com/apache/flink/pull/1227#discussion_r41241923","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/1227","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/41241923"},"html":{"href":"https://github.com/apache/flink/pull/1227#discussion_r41241923"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/1227"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/1227","id":46796105,"html_url":"https://github.com/apache/flink/pull/1227","diff_url":"https://github.com/apache/flink/pull/1227.diff","patch_url":"https://github.com/apache/flink/pull/1227.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/1227","number":1227,"state":"open","locked":false,"title":"[FLINK-2805] [blobmanager] Write JARs to file state backend for 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":"This is a follow up to #1153. I've taken two changes from #1153 for convenience. Other than that, this is independent.\r\n\r\nWhen running the `BlobServer` in `RecoveryMode#ZOOKEEPER`, this will upload the JARs to the configured file system backend (e.g. HDFS). **Important**: it introduces a hard dependency to have a configured file state backend when running the blob server with recovery. This is in line with #1153.\r\n\r\nThis JAR copying only happens on the server side, e.g. the client uploads to the server and the server uploads it to the state backend. Same when requesting a locally non-existing blob: the client requests from the server and the server downloads if not available and then answers the client.\r\n\r\nThere are other ways to implement this, but this one was minimally invasive and fully circumvents any Akka actor threads for downloading/uploading. A more invasive change could allow to directly interact with the state backend on the client side [task manager] as well. This would spread the load better across the cluster in case of a DFS and save unnecessary network transfers.","created_at":"2015-10-05T15:34:55Z","updated_at":"2015-10-06T09:16:11Z","closed_at":null,"merged_at":null,"merge_commit_sha":"d21cb932f4b9801d37f378d7a37ebd421eb7cd1c","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/1227/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/1227/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/1227/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/00ea6b02f15f00486c9541af08b8197c41dd94f7","head":{"label":"uce:hdfs_jars-2805","ref":"hdfs_jars-2805","sha":"00ea6b02f15f00486c9541af08b8197c41dd94f7","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-10-06T08:23:08Z","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":64016,"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":"9513f0e33f7aba46ebcc322d51ef12f0302ec2c2","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-05T15:49:25Z","pushed_at":"2015-10-06T09:10:15Z","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":2588927,"stargazers_count":657,"watchers_count":657,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":447,"mirror_url":"git://git.apache.org/incubator-flink.git","open_issues_count":81,"forks":447,"open_issues":81,"watchers":657,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/1227"},"html":{"href":"https://github.com/apache/flink/pull/1227"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/1227"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/1227/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/1227/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/1227/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/00ea6b02f15f00486c9541af08b8197c41dd94f7"}}}}