{"action":"created","comment":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/comments/30189167","id":30189167,"diff_hunk":"@@ -116,97 +132,69 @@ public AbstractJobLauncher(Properties properties)\n       taskExecutor.submit(task);\n     }\n \n-    LOG.info(String.format(\"Waiting for submitted tasks of job %s to complete...\", jobId));\n-    while (countDownLatch.getCount() > 0) {\n-      LOG.info(String\n-          .format(\"%d out of %d tasks of job %s are running\", countDownLatch.getCount(), workUnits.size(), jobId));\n-      countDownLatch.await(1, TimeUnit.MINUTES);\n-    }\n-    LOG.info(String.format(\"All tasks of job %s have completed\", jobId));\n-\n     return tasks;\n   }\n \n+  /**\n+   * This implementation simply sets a flag that indicate cancellation is requested and waits\n+   * for the cancellation to be executed. It does not guarantee successful job cancellation.\n+   *\n+   * {@inheritDoc JobLauncher#cancelJob(JobListener)}\n+   */\n   @Override\n-  @SuppressWarnings(\"unchecked\")\n-  public void launchJob(Properties jobProps, JobListener jobListener)\n+  public void cancelJob(JobListener jobListener)\n       throws JobException {\n-    Preconditions.checkNotNull(jobProps);\n-    Preconditions.checkArgument(\n-        jobProps.containsKey(ConfigurationKeys.JOB_NAME_KEY), \"A job must have a job name specified by job.name\");\n+    synchronized (this.cancellationExecutor) {\n+      if (this.cancellationRequested) {\n+        LOG.info(String.format(\"Job %s has already been cancelled or cancellation has already been requested\",\n+            this.jobContext.getJobId()));\n+        return;\n+      }\n \n-    String jobName = jobProps.getProperty(ConfigurationKeys.JOB_NAME_KEY);\n+      // Request cancellation\n+      this.cancellationRequested = true;\n \n-    if (Boolean.valueOf(jobProps.getProperty(ConfigurationKeys.JOB_DISABLED_KEY, Boolean.FALSE.toString()))) {\n-      LOG.info(String.format(\"Not launching job %s as it is disabled\", jobName));\n-      return;\n-    }\n-\n-    // Get the job lock if job locking is enabled\n-    Optional<JobLock> jobLockOptional = Optional.absent();\n-    boolean jobLockEnabled =\n-        Boolean.valueOf(jobProps.getProperty(ConfigurationKeys.JOB_LOCK_ENABLED_KEY, Boolean.TRUE.toString()));\n-    if (jobLockEnabled) {\n-      try {\n-        jobLockOptional = Optional.of(getJobLock(jobName, jobProps));\n-      } catch (IOException ioe) {\n-        throw new JobException(\"Failed to get job lock for job \" + jobName, ioe);\n+      // Wait for the cancellation to be executed\n+      synchronized (this.cancellationLock) {\n+        try {\n+          this.cancellationLock.wait(TimeUnit.SECONDS.toMillis(60));\n+        } catch (InterruptedException ie) {\n+          Thread.currentThread().interrupt();\n+          return;\n+        }\n       }\n     }\n \n-    // Try acquiring the job lock before proceeding\n-    if (!tryLockJob(jobName, jobLockOptional)) {\n-      throw new JobException(\n-          String.format(\"Previous instance of job %s is still running, skipping this scheduled run\", jobName));\n+    if (jobListener != null) {\n+      jobListener.onJobCancellation(this.jobContext.getJobState());\n     }\n+  }\n \n-    String jobId = jobProps.getProperty(ConfigurationKeys.JOB_ID_KEY);\n-    // If no job ID is assigned (e.g., if the job is assigned through Azkaban), assign a new job ID here.\n-    if (Strings.isNullOrEmpty(jobId)) {\n-      jobId = JobLauncherUtils.newJobId(jobName);\n-      jobProps.setProperty(ConfigurationKeys.JOB_ID_KEY, jobId);\n-    }\n-\n-    State jobPropsState = new State();\n-    // Add all job configuration properties of this job\n-    jobPropsState.addAll(jobProps);\n-\n-\n-    // Initialize the source for the job\n-    JobState jobState;\n-    Source<?, ?> source;\n-    try {\n-      JobState previousJobState = getPreviousJobState(jobName);\n-      jobState = new JobState(jobPropsState, previousJobState.getTaskStatesAsWorkUnitStates(), jobName, jobId);\n-      // Remember the number of consecutive failures of this job in the past\n-      jobState.setProp(ConfigurationKeys.JOB_FAILURES_KEY,\n-          previousJobState.getPropAsInt(ConfigurationKeys.JOB_FAILURES_KEY, 0));\n-      source = new SourceDecorator(initSource(jobProps), jobId, LOG);\n-    } catch (Throwable t) {\n-      String errMsg = \"Failed to initialize the source for job \" + jobId;\n-      LOG.error(errMsg + \": \" + t, t);\n-      unlockJob(jobName, jobLockOptional);\n-      throw new JobException(errMsg, t);\n+  @Override\n+  @SuppressWarnings(\"unchecked\")\n+  public void launchJob(JobListener jobListener)\n+      throws JobException {\n+    if (!tryLockJob()) {\n+      throw new JobException(\n+          String.format(\"Previous instance of job %s is still running, skipping this scheduled run\",\n+              this.jobContext.getJobName()));\n     }\n \n-    JobMetrics jobMetrics = JobMetrics.get(jobState);\n-    jobState.setProp(ConfigurationKeys.METRIC_CONTEXT_NAME_KEY, jobMetrics.getName());\n-    jobState.setState(JobState.RunningState.PENDING);\n+    String jobId = this.jobContext.getJobId();\n+    JobState jobState = this.jobContext.getJobState();\n \n     // Generate work units of the job from the source\n-    Optional<List<WorkUnit>> workUnits = Optional.fromNullable(source.getWorkunits(jobState));\n+    Optional<List<WorkUnit>> workUnits = Optional.fromNullable(this.jobContext.getSource().getWorkunits(jobState));\n+    // The absence means there is something wrong getting the work units\n     if (!workUnits.isPresent()) {\n-      // The absence means there is something wrong getting the work units\n-      source.shutdown(jobState);\n-      unlockJob(jobName, jobLockOptional);\n+      unlockJob();\n       throw new JobException(\"Failed to get work units for job \" + jobId);\n     }\n \n+    // No work unit to run\n     if (workUnits.get().isEmpty()) {\n-      // No real work to do\n       LOG.warn(\"No work units have been created for job \" + jobId);\n-      source.shutdown(jobState);\n-      unlockJob(jobName, jobLockOptional);\n+      unlockJob();","path":"gobblin-runtime/src/main/java/gobblin/runtime/AbstractJobLauncher.java","position":263,"original_position":263,"commit_id":"bb315aeb8041f30a6c7922be74a5d536d5b115ac","original_commit_id":"bb315aeb8041f30a6c7922be74a5d536d5b115ac","user":{"login":"chavdar","id":928795,"avatar_url":"https://avatars.githubusercontent.com/u/928795?v=3","gravatar_id":"","url":"https://api.github.com/users/chavdar","html_url":"https://github.com/chavdar","followers_url":"https://api.github.com/users/chavdar/followers","following_url":"https://api.github.com/users/chavdar/following{/other_user}","gists_url":"https://api.github.com/users/chavdar/gists{/gist_id}","starred_url":"https://api.github.com/users/chavdar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chavdar/subscriptions","organizations_url":"https://api.github.com/users/chavdar/orgs","repos_url":"https://api.github.com/users/chavdar/repos","events_url":"https://api.github.com/users/chavdar/events{/privacy}","received_events_url":"https://api.github.com/users/chavdar/received_events","type":"User","site_admin":false},"body":"unlockJob() should be in a finally section. Having to call unlockJob in every exit branch is very error-prone.","created_at":"2015-05-12T22:55:07Z","updated_at":"2015-05-12T22:55:07Z","html_url":"https://github.com/linkedin/gobblin/pull/120#discussion_r30189167","pull_request_url":"https://api.github.com/repos/linkedin/gobblin/pulls/120","_links":{"self":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/comments/30189167"},"html":{"href":"https://github.com/linkedin/gobblin/pull/120#discussion_r30189167"},"pull_request":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/120"}}},"pull_request":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/120","id":34660826,"html_url":"https://github.com/linkedin/gobblin/pull/120","diff_url":"https://github.com/linkedin/gobblin/pull/120.diff","patch_url":"https://github.com/linkedin/gobblin/pull/120.patch","issue_url":"https://api.github.com/repos/linkedin/gobblin/issues/120","number":120,"state":"open","locked":false,"title":"Code refactoring around JobLauncher (see details below)","user":{"login":"liyinan926","id":3709137,"avatar_url":"https://avatars.githubusercontent.com/u/3709137?v=3","gravatar_id":"","url":"https://api.github.com/users/liyinan926","html_url":"https://github.com/liyinan926","followers_url":"https://api.github.com/users/liyinan926/followers","following_url":"https://api.github.com/users/liyinan926/following{/other_user}","gists_url":"https://api.github.com/users/liyinan926/gists{/gist_id}","starred_url":"https://api.github.com/users/liyinan926/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/liyinan926/subscriptions","organizations_url":"https://api.github.com/users/liyinan926/orgs","repos_url":"https://api.github.com/users/liyinan926/repos","events_url":"https://api.github.com/users/liyinan926/events{/privacy}","received_events_url":"https://api.github.com/users/liyinan926/received_events","type":"User","site_admin":false},"body":"1. Refactored `JobLauncher` to extend `Closeable` and code around its usage to properly close it like other `Closeable`s.\r\n2. Made clear that `JobLauncher` should not be reused and refactored code to somehow reinforce it.\r\n3. Refactored implementation classes of `JobLauncher` to follow the above changes.\r\n4. In `MRJobLauncher`, the `close()` method will now kill the Hadoop job if it has not completed.\r\n5. Other code refactoring.","created_at":"2015-05-04T18:22:26Z","updated_at":"2015-05-12T22:55:07Z","closed_at":null,"merged_at":null,"merge_commit_sha":"24d2fa36202eedc9c6555fe6c2543dc3fb988acc","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/linkedin/gobblin/pulls/120/commits","review_comments_url":"https://api.github.com/repos/linkedin/gobblin/pulls/120/comments","review_comment_url":"https://api.github.com/repos/linkedin/gobblin/pulls/comments{/number}","comments_url":"https://api.github.com/repos/linkedin/gobblin/issues/120/comments","statuses_url":"https://api.github.com/repos/linkedin/gobblin/statuses/bb315aeb8041f30a6c7922be74a5d536d5b115ac","head":{"label":"liyinan926:closeable-launcher","ref":"closeable-launcher","sha":"bb315aeb8041f30a6c7922be74a5d536d5b115ac","user":{"login":"liyinan926","id":3709137,"avatar_url":"https://avatars.githubusercontent.com/u/3709137?v=3","gravatar_id":"","url":"https://api.github.com/users/liyinan926","html_url":"https://github.com/liyinan926","followers_url":"https://api.github.com/users/liyinan926/followers","following_url":"https://api.github.com/users/liyinan926/following{/other_user}","gists_url":"https://api.github.com/users/liyinan926/gists{/gist_id}","starred_url":"https://api.github.com/users/liyinan926/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/liyinan926/subscriptions","organizations_url":"https://api.github.com/users/liyinan926/orgs","repos_url":"https://api.github.com/users/liyinan926/repos","events_url":"https://api.github.com/users/liyinan926/events{/privacy}","received_events_url":"https://api.github.com/users/liyinan926/received_events","type":"User","site_admin":false},"repo":{"id":31117112,"name":"gobblin","full_name":"liyinan926/gobblin","owner":{"login":"liyinan926","id":3709137,"avatar_url":"https://avatars.githubusercontent.com/u/3709137?v=3","gravatar_id":"","url":"https://api.github.com/users/liyinan926","html_url":"https://github.com/liyinan926","followers_url":"https://api.github.com/users/liyinan926/followers","following_url":"https://api.github.com/users/liyinan926/following{/other_user}","gists_url":"https://api.github.com/users/liyinan926/gists{/gist_id}","starred_url":"https://api.github.com/users/liyinan926/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/liyinan926/subscriptions","organizations_url":"https://api.github.com/users/liyinan926/orgs","repos_url":"https://api.github.com/users/liyinan926/repos","events_url":"https://api.github.com/users/liyinan926/events{/privacy}","received_events_url":"https://api.github.com/users/liyinan926/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/liyinan926/gobblin","description":"","fork":true,"url":"https://api.github.com/repos/liyinan926/gobblin","forks_url":"https://api.github.com/repos/liyinan926/gobblin/forks","keys_url":"https://api.github.com/repos/liyinan926/gobblin/keys{/key_id}","collaborators_url":"https://api.github.com/repos/liyinan926/gobblin/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/liyinan926/gobblin/teams","hooks_url":"https://api.github.com/repos/liyinan926/gobblin/hooks","issue_events_url":"https://api.github.com/repos/liyinan926/gobblin/issues/events{/number}","events_url":"https://api.github.com/repos/liyinan926/gobblin/events","assignees_url":"https://api.github.com/repos/liyinan926/gobblin/assignees{/user}","branches_url":"https://api.github.com/repos/liyinan926/gobblin/branches{/branch}","tags_url":"https://api.github.com/repos/liyinan926/gobblin/tags","blobs_url":"https://api.github.com/repos/liyinan926/gobblin/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/liyinan926/gobblin/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/liyinan926/gobblin/git/refs{/sha}","trees_url":"https://api.github.com/repos/liyinan926/gobblin/git/trees{/sha}","statuses_url":"https://api.github.com/repos/liyinan926/gobblin/statuses/{sha}","languages_url":"https://api.github.com/repos/liyinan926/gobblin/languages","stargazers_url":"https://api.github.com/repos/liyinan926/gobblin/stargazers","contributors_url":"https://api.github.com/repos/liyinan926/gobblin/contributors","subscribers_url":"https://api.github.com/repos/liyinan926/gobblin/subscribers","subscription_url":"https://api.github.com/repos/liyinan926/gobblin/subscription","commits_url":"https://api.github.com/repos/liyinan926/gobblin/commits{/sha}","git_commits_url":"https://api.github.com/repos/liyinan926/gobblin/git/commits{/sha}","comments_url":"https://api.github.com/repos/liyinan926/gobblin/comments{/number}","issue_comment_url":"https://api.github.com/repos/liyinan926/gobblin/issues/comments{/number}","contents_url":"https://api.github.com/repos/liyinan926/gobblin/contents/{+path}","compare_url":"https://api.github.com/repos/liyinan926/gobblin/compare/{base}...{head}","merges_url":"https://api.github.com/repos/liyinan926/gobblin/merges","archive_url":"https://api.github.com/repos/liyinan926/gobblin/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/liyinan926/gobblin/downloads","issues_url":"https://api.github.com/repos/liyinan926/gobblin/issues{/number}","pulls_url":"https://api.github.com/repos/liyinan926/gobblin/pulls{/number}","milestones_url":"https://api.github.com/repos/liyinan926/gobblin/milestones{/number}","notifications_url":"https://api.github.com/repos/liyinan926/gobblin/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/liyinan926/gobblin/labels{/name}","releases_url":"https://api.github.com/repos/liyinan926/gobblin/releases{/id}","created_at":"2015-02-21T06:07:55Z","updated_at":"2015-05-05T00:31:33Z","pushed_at":"2015-05-12T19:22:07Z","git_url":"git://github.com/liyinan926/gobblin.git","ssh_url":"git@github.com:liyinan926/gobblin.git","clone_url":"https://github.com/liyinan926/gobblin.git","svn_url":"https://github.com/liyinan926/gobblin","homepage":null,"size":77238,"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":"f5bebda6ba3b624f9c451d9fcb162cdccad37633","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-05-07T20:00:41Z","pushed_at":"2015-05-12T21:54:19Z","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":157360,"stargazers_count":37,"watchers_count":37,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":24,"mirror_url":null,"open_issues_count":7,"forks":24,"open_issues":7,"watchers":37,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/120"},"html":{"href":"https://github.com/linkedin/gobblin/pull/120"},"issue":{"href":"https://api.github.com/repos/linkedin/gobblin/issues/120"},"comments":{"href":"https://api.github.com/repos/linkedin/gobblin/issues/120/comments"},"review_comments":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/120/comments"},"review_comment":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/120/commits"},"statuses":{"href":"https://api.github.com/repos/linkedin/gobblin/statuses/bb315aeb8041f30a6c7922be74a5d536d5b115ac"}}}}