{"action":"created","comment":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/comments/26092448","id":26092448,"diff_hunk":"@@ -0,0 +1,151 @@\n+package gobblin.runtime.mapreduce;\n+\n+import gobblin.configuration.ConfigurationKeys;\n+import gobblin.source.workunit.WorkUnit;\n+import gobblin.util.ForkOperatorUtils;\n+\n+import java.io.DataInputStream;\n+import java.io.IOException;\n+import java.net.URI;\n+\n+import org.apache.hadoop.conf.Configuration;\n+import org.apache.hadoop.fs.FileStatus;\n+import org.apache.hadoop.fs.FileSystem;\n+import org.apache.hadoop.fs.Path;\n+import org.apache.hadoop.io.NullWritable;\n+import org.apache.hadoop.mapreduce.JobContext;\n+import org.apache.hadoop.mapreduce.JobStatus;\n+import org.apache.hadoop.mapreduce.OutputCommitter;\n+import org.apache.hadoop.mapreduce.TaskAttemptContext;\n+import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;\n+\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import com.google.common.io.Closer;\n+\n+/**\n+ * Hadoop {@link org.apache.hadoop.mapreduce.OutputFormat} implementation with a custom {@link OutputCommitter}\n+ */\n+public class GobblinOutputFormat extends NullOutputFormat<NullWritable, NullWritable> {\n+\n+  private static final Logger LOG = LoggerFactory.getLogger(GobblinOutputFormat.class);\n+\n+  @Override\n+  public OutputCommitter getOutputCommitter(TaskAttemptContext context) {\n+    return new GobblinOutputCommitter();\n+  }\n+\n+  /**\n+   * Hadoop {@link OutputCommitter} implementation that overrides the default\n+   * {@link #abortJob(JobContext, org.apache.hadoop.mapreduce.JobStatus.State)} behavior. This is necessary to add\n+   * functionality for cleaning up staging data when the\n+   * {@link com.linkedin.uif.runtime.JobLauncher#cancelJob(Properties)} method is called via Azkaban. Azkaban only\n+   * allows the cancel method run to for 5 ms until it does a hard kill on the process, so instead, we will clean up the\n+   * staging data in the AM.\n+   */\n+  private class GobblinOutputCommitter extends OutputCommitter {\n+\n+    @Override\n+    public void abortJob(JobContext jobContext, JobStatus.State state) throws IOException {\n+      LOG.info(\"Aborting Job: \" + jobContext.getJobID());\n+\n+      Configuration conf = jobContext.getConfiguration();\n+\n+      URI fsUri = URI.create(conf.get(ConfigurationKeys.FS_URI_KEY, ConfigurationKeys.LOCAL_FS_URI));\n+      FileSystem fs = FileSystem.get(fsUri, conf);\n+\n+      Path mrJobDir =\n+          new Path(conf.get(ConfigurationKeys.MR_JOB_ROOT_DIR_KEY), conf.get(ConfigurationKeys.JOB_NAME_KEY));\n+      Path jobInputDir = new Path(mrJobDir, \"input\");\n+\n+      for (FileStatus status : fs.listStatus(jobInputDir)) {\n+\n+        Closer workUnitFileCloser = Closer.create();\n+        if (status.getPath().getName().endsWith(\".wu\")) {\n+\n+          WorkUnit wu = new WorkUnit();\n+          wu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath()))));","path":"gobblin-runtime/src/main/java/gobblin/runtime/mapreduce/GobblinOutputFormat.java","position":68,"original_position":68,"commit_id":"d4cb183f18856d4788bb91f67b13c54a2481768f","original_commit_id":"d4cb183f18856d4788bb91f67b13c54a2481768f","user":{"login":"sahilTakiar","id":1282583,"avatar_url":"https://avatars.githubusercontent.com/u/1282583?v=3","gravatar_id":"","url":"https://api.github.com/users/sahilTakiar","html_url":"https://github.com/sahilTakiar","followers_url":"https://api.github.com/users/sahilTakiar/followers","following_url":"https://api.github.com/users/sahilTakiar/following{/other_user}","gists_url":"https://api.github.com/users/sahilTakiar/gists{/gist_id}","starred_url":"https://api.github.com/users/sahilTakiar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sahilTakiar/subscriptions","organizations_url":"https://api.github.com/users/sahilTakiar/orgs","repos_url":"https://api.github.com/users/sahilTakiar/repos","events_url":"https://api.github.com/users/sahilTakiar/events{/privacy}","received_events_url":"https://api.github.com/users/sahilTakiar/received_events","type":"User","site_admin":false},"body":"I can't for a DataInputStream, also its just a serialized WorkUnit object, not a text file.","created_at":"2015-03-10T01:13:50Z","updated_at":"2015-03-10T01:13:50Z","html_url":"https://github.com/linkedin/gobblin/pull/44#discussion_r26092448","pull_request_url":"https://api.github.com/repos/linkedin/gobblin/pulls/44","_links":{"self":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/comments/26092448"},"html":{"href":"https://github.com/linkedin/gobblin/pull/44#discussion_r26092448"},"pull_request":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/44"}}},"pull_request":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/44","id":30685184,"html_url":"https://github.com/linkedin/gobblin/pull/44","diff_url":"https://github.com/linkedin/gobblin/pull/44.diff","patch_url":"https://github.com/linkedin/gobblin/pull/44.patch","issue_url":"https://api.github.com/repos/linkedin/gobblin/issues/44","number":44,"state":"open","locked":false,"title":"Adding support for cleaning up staging data when Gobblin is cancelled via the Azkaban UI","user":{"login":"sahilTakiar","id":1282583,"avatar_url":"https://avatars.githubusercontent.com/u/1282583?v=3","gravatar_id":"","url":"https://api.github.com/users/sahilTakiar","html_url":"https://github.com/sahilTakiar","followers_url":"https://api.github.com/users/sahilTakiar/followers","following_url":"https://api.github.com/users/sahilTakiar/following{/other_user}","gists_url":"https://api.github.com/users/sahilTakiar/gists{/gist_id}","starred_url":"https://api.github.com/users/sahilTakiar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sahilTakiar/subscriptions","organizations_url":"https://api.github.com/users/sahilTakiar/orgs","repos_url":"https://api.github.com/users/sahilTakiar/repos","events_url":"https://api.github.com/users/sahilTakiar/events{/privacy}","received_events_url":"https://api.github.com/users/sahilTakiar/received_events","type":"User","site_admin":false},"body":"Here is the first version. I still need to do some cleanup, add some comments, and write a test. Making this pull request to see what everyone thinks of the approach.\r\n\r\nAzkaban only lets the cancel method run for 5 ms, before doing a hard kill on the process. This isn't enough time to cleanup staging data, but should be enough time to kill the Hadoop job (at least based on historical behavior). The OutputCommitter allows the user to hook into the lifecycle of the task and the jobs, so when killJob is called by the client, the AM will call OutputCommitter.abortJob() method. I have over--ridden this method, and added logic to clean up the staging data. Since the AM is in a different process than the client (running on Azkaban), the code de-serializes the WorkUnit files written to HDFS in order to obtain the writer output folder locations. Once it constructs the writer output paths, it deletes them.\r\n\r\n@liyinan926 what do you think of this approach?","created_at":"2015-03-06T22:24:44Z","updated_at":"2015-03-10T01:13:50Z","closed_at":null,"merged_at":null,"merge_commit_sha":"7c1f2d66f4cb54ccb342168d58de2f9c4d6ee784","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/linkedin/gobblin/pulls/44/commits","review_comments_url":"https://api.github.com/repos/linkedin/gobblin/pulls/44/comments","review_comment_url":"https://api.github.com/repos/linkedin/gobblin/pulls/comments{/number}","comments_url":"https://api.github.com/repos/linkedin/gobblin/issues/44/comments","statuses_url":"https://api.github.com/repos/linkedin/gobblin/statuses/d4cb183f18856d4788bb91f67b13c54a2481768f","head":{"label":"sahilTakiar:master","ref":"master","sha":"d4cb183f18856d4788bb91f67b13c54a2481768f","user":{"login":"sahilTakiar","id":1282583,"avatar_url":"https://avatars.githubusercontent.com/u/1282583?v=3","gravatar_id":"","url":"https://api.github.com/users/sahilTakiar","html_url":"https://github.com/sahilTakiar","followers_url":"https://api.github.com/users/sahilTakiar/followers","following_url":"https://api.github.com/users/sahilTakiar/following{/other_user}","gists_url":"https://api.github.com/users/sahilTakiar/gists{/gist_id}","starred_url":"https://api.github.com/users/sahilTakiar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sahilTakiar/subscriptions","organizations_url":"https://api.github.com/users/sahilTakiar/orgs","repos_url":"https://api.github.com/users/sahilTakiar/repos","events_url":"https://api.github.com/users/sahilTakiar/events{/privacy}","received_events_url":"https://api.github.com/users/sahilTakiar/received_events","type":"User","site_admin":false},"repo":{"id":31343048,"name":"gobblin","full_name":"sahilTakiar/gobblin","owner":{"login":"sahilTakiar","id":1282583,"avatar_url":"https://avatars.githubusercontent.com/u/1282583?v=3","gravatar_id":"","url":"https://api.github.com/users/sahilTakiar","html_url":"https://github.com/sahilTakiar","followers_url":"https://api.github.com/users/sahilTakiar/followers","following_url":"https://api.github.com/users/sahilTakiar/following{/other_user}","gists_url":"https://api.github.com/users/sahilTakiar/gists{/gist_id}","starred_url":"https://api.github.com/users/sahilTakiar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sahilTakiar/subscriptions","organizations_url":"https://api.github.com/users/sahilTakiar/orgs","repos_url":"https://api.github.com/users/sahilTakiar/repos","events_url":"https://api.github.com/users/sahilTakiar/events{/privacy}","received_events_url":"https://api.github.com/users/sahilTakiar/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/sahilTakiar/gobblin","description":"","fork":true,"url":"https://api.github.com/repos/sahilTakiar/gobblin","forks_url":"https://api.github.com/repos/sahilTakiar/gobblin/forks","keys_url":"https://api.github.com/repos/sahilTakiar/gobblin/keys{/key_id}","collaborators_url":"https://api.github.com/repos/sahilTakiar/gobblin/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/sahilTakiar/gobblin/teams","hooks_url":"https://api.github.com/repos/sahilTakiar/gobblin/hooks","issue_events_url":"https://api.github.com/repos/sahilTakiar/gobblin/issues/events{/number}","events_url":"https://api.github.com/repos/sahilTakiar/gobblin/events","assignees_url":"https://api.github.com/repos/sahilTakiar/gobblin/assignees{/user}","branches_url":"https://api.github.com/repos/sahilTakiar/gobblin/branches{/branch}","tags_url":"https://api.github.com/repos/sahilTakiar/gobblin/tags","blobs_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/refs{/sha}","trees_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/trees{/sha}","statuses_url":"https://api.github.com/repos/sahilTakiar/gobblin/statuses/{sha}","languages_url":"https://api.github.com/repos/sahilTakiar/gobblin/languages","stargazers_url":"https://api.github.com/repos/sahilTakiar/gobblin/stargazers","contributors_url":"https://api.github.com/repos/sahilTakiar/gobblin/contributors","subscribers_url":"https://api.github.com/repos/sahilTakiar/gobblin/subscribers","subscription_url":"https://api.github.com/repos/sahilTakiar/gobblin/subscription","commits_url":"https://api.github.com/repos/sahilTakiar/gobblin/commits{/sha}","git_commits_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/commits{/sha}","comments_url":"https://api.github.com/repos/sahilTakiar/gobblin/comments{/number}","issue_comment_url":"https://api.github.com/repos/sahilTakiar/gobblin/issues/comments{/number}","contents_url":"https://api.github.com/repos/sahilTakiar/gobblin/contents/{+path}","compare_url":"https://api.github.com/repos/sahilTakiar/gobblin/compare/{base}...{head}","merges_url":"https://api.github.com/repos/sahilTakiar/gobblin/merges","archive_url":"https://api.github.com/repos/sahilTakiar/gobblin/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/sahilTakiar/gobblin/downloads","issues_url":"https://api.github.com/repos/sahilTakiar/gobblin/issues{/number}","pulls_url":"https://api.github.com/repos/sahilTakiar/gobblin/pulls{/number}","milestones_url":"https://api.github.com/repos/sahilTakiar/gobblin/milestones{/number}","notifications_url":"https://api.github.com/repos/sahilTakiar/gobblin/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/sahilTakiar/gobblin/labels{/name}","releases_url":"https://api.github.com/repos/sahilTakiar/gobblin/releases{/id}","created_at":"2015-02-26T00:28:16Z","updated_at":"2015-03-06T22:05:36Z","pushed_at":"2015-03-06T22:05:36Z","git_url":"git://github.com/sahilTakiar/gobblin.git","ssh_url":"git@github.com:sahilTakiar/gobblin.git","clone_url":"https://github.com/sahilTakiar/gobblin.git","svn_url":"https://github.com/sahilTakiar/gobblin","homepage":null,"size":75981,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"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":"linkedin:master","ref":"master","sha":"1175d1b8a09c41d37dec74d895a287fd9172ad15","user":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"repo":{"id":27392543,"name":"gobblin","full_name":"linkedin/gobblin","owner":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/linkedin/gobblin","description":"","fork":false,"url":"https://api.github.com/repos/linkedin/gobblin","forks_url":"https://api.github.com/repos/linkedin/gobblin/forks","keys_url":"https://api.github.com/repos/linkedin/gobblin/keys{/key_id}","collaborators_url":"https://api.github.com/repos/linkedin/gobblin/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/linkedin/gobblin/teams","hooks_url":"https://api.github.com/repos/linkedin/gobblin/hooks","issue_events_url":"https://api.github.com/repos/linkedin/gobblin/issues/events{/number}","events_url":"https://api.github.com/repos/linkedin/gobblin/events","assignees_url":"https://api.github.com/repos/linkedin/gobblin/assignees{/user}","branches_url":"https://api.github.com/repos/linkedin/gobblin/branches{/branch}","tags_url":"https://api.github.com/repos/linkedin/gobblin/tags","blobs_url":"https://api.github.com/repos/linkedin/gobblin/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/linkedin/gobblin/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/linkedin/gobblin/git/refs{/sha}","trees_url":"https://api.github.com/repos/linkedin/gobblin/git/trees{/sha}","statuses_url":"https://api.github.com/repos/linkedin/gobblin/statuses/{sha}","languages_url":"https://api.github.com/repos/linkedin/gobblin/languages","stargazers_url":"https://api.github.com/repos/linkedin/gobblin/stargazers","contributors_url":"https://api.github.com/repos/linkedin/gobblin/contributors","subscribers_url":"https://api.github.com/repos/linkedin/gobblin/subscribers","subscription_url":"https://api.github.com/repos/linkedin/gobblin/subscription","commits_url":"https://api.github.com/repos/linkedin/gobblin/commits{/sha}","git_commits_url":"https://api.github.com/repos/linkedin/gobblin/git/commits{/sha}","comments_url":"https://api.github.com/repos/linkedin/gobblin/comments{/number}","issue_comment_url":"https://api.github.com/repos/linkedin/gobblin/issues/comments{/number}","contents_url":"https://api.github.com/repos/linkedin/gobblin/contents/{+path}","compare_url":"https://api.github.com/repos/linkedin/gobblin/compare/{base}...{head}","merges_url":"https://api.github.com/repos/linkedin/gobblin/merges","archive_url":"https://api.github.com/repos/linkedin/gobblin/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/linkedin/gobblin/downloads","issues_url":"https://api.github.com/repos/linkedin/gobblin/issues{/number}","pulls_url":"https://api.github.com/repos/linkedin/gobblin/pulls{/number}","milestones_url":"https://api.github.com/repos/linkedin/gobblin/milestones{/number}","notifications_url":"https://api.github.com/repos/linkedin/gobblin/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/linkedin/gobblin/labels{/name}","releases_url":"https://api.github.com/repos/linkedin/gobblin/releases{/id}","created_at":"2014-12-01T18:10:50Z","updated_at":"2015-03-09T20:17:02Z","pushed_at":"2015-03-09T20:17:00Z","git_url":"git://github.com/linkedin/gobblin.git","ssh_url":"git@github.com:linkedin/gobblin.git","clone_url":"https://github.com/linkedin/gobblin.git","svn_url":"https://github.com/linkedin/gobblin","homepage":null,"size":151488,"stargazers_count":11,"watchers_count":11,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":9,"mirror_url":null,"open_issues_count":6,"forks":9,"open_issues":6,"watchers":11,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/44"},"html":{"href":"https://github.com/linkedin/gobblin/pull/44"},"issue":{"href":"https://api.github.com/repos/linkedin/gobblin/issues/44"},"comments":{"href":"https://api.github.com/repos/linkedin/gobblin/issues/44/comments"},"review_comments":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/44/comments"},"review_comment":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/44/commits"},"statuses":{"href":"https://api.github.com/repos/linkedin/gobblin/statuses/d4cb183f18856d4788bb91f67b13c54a2481768f"}}}}