@Test public void testRunWithTaskCreationFailure() throws Exception { WorkUnit tmpWU = new WorkUnit(); tmpWU.setProp(ConfigurationKeys.TASK_ID_KEY, "task_test"); List<WorkUnit> workUnit = ImmutableList.of(tmpWU); JobState jobState = new JobState(); jobState.setProp(RETRY_TIME_OUT_MS, 1000); TaskStateTracker stateTrackerMock = Mockito.mock(TaskStateTracker.class); taskAttempt = new GobblinMultiTaskAttempt(workUnit.iterator(), "testJob", jobState, stateTrackerMock, taskExecutorMock, Optional.absent(), Optional.absent(), jobBroker); try { taskAttempt.run(); } catch (Exception e) { Assert.assertTrue(e instanceof TaskCreationException); return; } Assert.fail(); }
public void run() throws IOException, InterruptedException { if (!this.workUnits.hasNext()) { log.warn("No work units to run in container " + containerIdOptional.or("")); return; } CountUpAndDownLatch countDownLatch = new CountUpAndDownLatch(0); Pair<List<Task>, Boolean> executionResult = runWorkUnits(countDownLatch); this.tasks = executionResult.getFirst(); if (!executionResult.getSecond()) { throw new TaskCreationException("Failing in submitting at least one task before execution."); } log.info("Waiting for submitted tasks of job {} to complete in container {}...", jobId, containerIdOptional.or("")); try { while (countDownLatch.getCount() > 0) { if (this.interruptionPredicate.test(this)) { log.info("Interrupting task execution due to satisfied predicate."); interruptTaskExecution(countDownLatch); break; } log.info(String.format("%d out of %d tasks of job %s are running in container %s", countDownLatch.getCount(), countDownLatch.getRegisteredParties(), jobId, containerIdOptional.or(""))); if (countDownLatch.await(10, TimeUnit.SECONDS)) { break; } } } catch (InterruptedException interrupt) { log.info("Job interrupted by InterrupedException."); interruptTaskExecution(countDownLatch); } log.info("All assigned tasks of job {} have completed in container {}", jobId, containerIdOptional.or("")); }
GobblinMultiTaskAttempt { public void run() throws IOException, InterruptedException { if (!this.workUnits.hasNext()) { log.warn("No work units to run in container " + containerIdOptional.or("")); return; } CountUpAndDownLatch countDownLatch = new CountUpAndDownLatch(0); Pair<List<Task>, Boolean> executionResult = runWorkUnits(countDownLatch); this.tasks = executionResult.getFirst(); if (!executionResult.getSecond()) { throw new TaskCreationException("Failing in submitting at least one task before execution."); } log.info("Waiting for submitted tasks of job {} to complete in container {}...", jobId, containerIdOptional.or("")); try { while (countDownLatch.getCount() > 0) { if (this.interruptionPredicate.test(this)) { log.info("Interrupting task execution due to satisfied predicate."); interruptTaskExecution(countDownLatch); break; } log.info(String.format("%d out of %d tasks of job %s are running in container %s", countDownLatch.getCount(), countDownLatch.getRegisteredParties(), jobId, containerIdOptional.or(""))); if (countDownLatch.await(10, TimeUnit.SECONDS)) { break; } } } catch (InterruptedException interrupt) { log.info("Job interrupted by InterrupedException."); interruptTaskExecution(countDownLatch); } log.info("All assigned tasks of job {} have completed in container {}", jobId, containerIdOptional.or("")); } }
GobblinMultiTaskAttempt { public void run() throws IOException, InterruptedException { if (!this.workUnits.hasNext()) { log.warn("No work units to run in container " + containerIdOptional.or("")); return; } CountUpAndDownLatch countDownLatch = new CountUpAndDownLatch(0); Pair<List<Task>, Boolean> executionResult = runWorkUnits(countDownLatch); this.tasks = executionResult.getFirst(); if (!executionResult.getSecond()) { throw new TaskCreationException("Failing in submitting at least one task before execution."); } log.info("Waiting for submitted tasks of job {} to complete in container {}...", jobId, containerIdOptional.or("")); try { while (countDownLatch.getCount() > 0) { if (this.interruptionPredicate.test(this)) { log.info("Interrupting task execution due to satisfied predicate."); interruptTaskExecution(countDownLatch); break; } log.info(String.format("%d out of %d tasks of job %s are running in container %s", countDownLatch.getCount(), countDownLatch.getRegisteredParties(), jobId, containerIdOptional.or(""))); if (countDownLatch.await(10, TimeUnit.SECONDS)) { break; } } } catch (InterruptedException interrupt) { log.info("Job interrupted by InterrupedException."); interruptTaskExecution(countDownLatch); } log.info("All assigned tasks of job {} have completed in container {}", jobId, containerIdOptional.or("")); } GobblinMultiTaskAttempt(Iterator<WorkUnit> workUnits, String jobId, JobState jobState,
      TaskStateTracker taskStateTracker, TaskExecutor taskExecutor, Optional<String> containerIdOptional,
      Optional<StateStore<TaskState>> taskStateStoreOptional, SharedResourcesBroker<GobblinScopeTypes> jobBroker); }
GobblinMultiTaskAttempt { public void run() throws IOException, InterruptedException { if (!this.workUnits.hasNext()) { log.warn("No work units to run in container " + containerIdOptional.or("")); return; } CountUpAndDownLatch countDownLatch = new CountUpAndDownLatch(0); Pair<List<Task>, Boolean> executionResult = runWorkUnits(countDownLatch); this.tasks = executionResult.getFirst(); if (!executionResult.getSecond()) { throw new TaskCreationException("Failing in submitting at least one task before execution."); } log.info("Waiting for submitted tasks of job {} to complete in container {}...", jobId, containerIdOptional.or("")); try { while (countDownLatch.getCount() > 0) { if (this.interruptionPredicate.test(this)) { log.info("Interrupting task execution due to satisfied predicate."); interruptTaskExecution(countDownLatch); break; } log.info(String.format("%d out of %d tasks of job %s are running in container %s", countDownLatch.getCount(), countDownLatch.getRegisteredParties(), jobId, containerIdOptional.or(""))); if (countDownLatch.await(10, TimeUnit.SECONDS)) { break; } } } catch (InterruptedException interrupt) { log.info("Job interrupted by InterrupedException."); interruptTaskExecution(countDownLatch); } log.info("All assigned tasks of job {} have completed in container {}", jobId, containerIdOptional.or("")); } GobblinMultiTaskAttempt(Iterator<WorkUnit> workUnits, String jobId, JobState jobState,
      TaskStateTracker taskStateTracker, TaskExecutor taskExecutor, Optional<String> containerIdOptional,
      Optional<StateStore<TaskState>> taskStateStoreOptional, SharedResourcesBroker<GobblinScopeTypes> jobBroker); void run(); void commit(); void shutdownTasks(); boolean isSpeculativeExecutionSafe(); void addCleanupCommitStep(CommitStep commitStep); void runAndOptionallyCommitTaskAttempt(CommitPolicy multiTaskAttemptCommitPolicy); void cleanMetrics(); static GobblinMultiTaskAttempt runWorkUnits(JobContext jobContext, Iterator<WorkUnit> workUnits,
      TaskStateTracker taskStateTracker, TaskExecutor taskExecutor, CommitPolicy multiTaskAttemptCommitPolicy); static GobblinMultiTaskAttempt runWorkUnits(String jobId, String containerId, JobState jobState,
      List<WorkUnit> workUnits, TaskStateTracker taskStateTracker, TaskExecutor taskExecutor,
      StateStore<TaskState> taskStateStore, CommitPolicy multiTaskAttemptCommitPolicy,
      SharedResourcesBroker<GobblinScopeTypes> jobBroker, Predicate<GobblinMultiTaskAttempt> interruptionPredicate); }
GobblinMultiTaskAttempt { public void run() throws IOException, InterruptedException { if (!this.workUnits.hasNext()) { log.warn("No work units to run in container " + containerIdOptional.or("")); return; } CountUpAndDownLatch countDownLatch = new CountUpAndDownLatch(0); Pair<List<Task>, Boolean> executionResult = runWorkUnits(countDownLatch); this.tasks = executionResult.getFirst(); if (!executionResult.getSecond()) { throw new TaskCreationException("Failing in submitting at least one task before execution."); } log.info("Waiting for submitted tasks of job {} to complete in container {}...", jobId, containerIdOptional.or("")); try { while (countDownLatch.getCount() > 0) { if (this.interruptionPredicate.test(this)) { log.info("Interrupting task execution due to satisfied predicate."); interruptTaskExecution(countDownLatch); break; } log.info(String.format("%d out of %d tasks of job %s are running in container %s", countDownLatch.getCount(), countDownLatch.getRegisteredParties(), jobId, containerIdOptional.or(""))); if (countDownLatch.await(10, TimeUnit.SECONDS)) { break; } } } catch (InterruptedException interrupt) { log.info("Job interrupted by InterrupedException."); interruptTaskExecution(countDownLatch); } log.info("All assigned tasks of job {} have completed in container {}", jobId, containerIdOptional.or("")); } GobblinMultiTaskAttempt(Iterator<WorkUnit> workUnits, String jobId, JobState jobState,
      TaskStateTracker taskStateTracker, TaskExecutor taskExecutor, Optional<String> containerIdOptional,
      Optional<StateStore<TaskState>> taskStateStoreOptional, SharedResourcesBroker<GobblinScopeTypes> jobBroker); void run(); void commit(); void shutdownTasks(); boolean isSpeculativeExecutionSafe(); void addCleanupCommitStep(CommitStep commitStep); void runAndOptionallyCommitTaskAttempt(CommitPolicy multiTaskAttemptCommitPolicy); void cleanMetrics(); static GobblinMultiTaskAttempt runWorkUnits(JobContext jobContext, Iterator<WorkUnit> workUnits,
      TaskStateTracker taskStateTracker, TaskExecutor taskExecutor, CommitPolicy multiTaskAttemptCommitPolicy); static GobblinMultiTaskAttempt runWorkUnits(String jobId, String containerId, JobState jobState,
      List<WorkUnit> workUnits, TaskStateTracker taskStateTracker, TaskExecutor taskExecutor,
      StateStore<TaskState> taskStateStore, CommitPolicy multiTaskAttemptCommitPolicy,
      SharedResourcesBroker<GobblinScopeTypes> jobBroker, Predicate<GobblinMultiTaskAttempt> interruptionPredicate);  }