@Test public void testPrepareTask() throws IOException, InterruptedException { EventBus eventBus = EventBusFactory.get(ContainerHealthCheckFailureEvent.CONTAINER_HEALTH_CHECK_EVENT_BUS_NAME, SharedResourcesBrokerFactory.getImplicitBroker()); eventBus.register(this); countDownLatchForFailInTaskCreation = new CountDownLatch(1); Path jobStateFilePath = new Path(appWorkDir, TestHelper.TEST_JOB_ID + "." + AbstractJobLauncher.JOB_STATE_FILE_NAME); JobState jobState = new JobState(); jobState.setJobName(TestHelper.TEST_JOB_NAME); jobState.setJobId(TestHelper.TEST_JOB_ID); SerializationUtils.serializeState(this.localFs, jobStateFilePath, jobState); WorkUnit workUnit = WorkUnit.createEmpty(); prepareWorkUnit(workUnit); File sourceJsonFile = new File(this.appWorkDir.toString(), TestHelper.TEST_JOB_NAME + ".json"); TestHelper.createSourceJsonFile(sourceJsonFile); workUnit.setProp(SimpleJsonSource.SOURCE_FILE_KEY, sourceJsonFile.getAbsolutePath()); Path workUnitDirPath = new Path(this.appWorkDir, GobblinClusterConfigurationKeys.INPUT_WORK_UNIT_DIR_NAME); FsStateStore<WorkUnit> wuStateStore = new FsStateStore<>(this.localFs, workUnitDirPath.toString(), WorkUnit.class); Path workUnitFilePath = new Path(new Path(workUnitDirPath, TestHelper.TEST_JOB_ID), TestHelper.TEST_JOB_NAME + ".wu"); wuStateStore.put(TestHelper.TEST_JOB_ID, TestHelper.TEST_JOB_NAME + ".wu", workUnit); Assert.assertTrue(this.localFs.exists(workUnitFilePath)); Map<String, String> taskConfigMap = Maps.newHashMap(); taskConfigMap.put(GobblinClusterConfigurationKeys.WORK_UNIT_FILE_PATH, workUnitFilePath.toString()); taskConfigMap.put(ConfigurationKeys.JOB_NAME_KEY, TestHelper.TEST_JOB_NAME); taskConfigMap.put(ConfigurationKeys.JOB_ID_KEY, TestHelper.TEST_JOB_ID); taskConfigMap.put(ConfigurationKeys.TASK_KEY_KEY, Long.toString(Id.parse(TestHelper.TEST_JOB_ID).getSequence())); TaskConfig taskConfig = new TaskConfig("", taskConfigMap, true); TaskCallbackContext taskCallbackContext = Mockito.mock(TaskCallbackContext.class); when(taskCallbackContext.getTaskConfig()).thenReturn(taskConfig); when(taskCallbackContext.getManager()).thenReturn(this.helixManager); TaskDriver taskDriver = createTaskDriverWithMockedAttributes(taskCallbackContext, taskConfig); TaskRunnerSuiteBase.Builder builder = new TaskRunnerSuiteBase.Builder(ConfigFactory.empty() .withValue(RETRY_TYPE, ConfigValueFactory.fromAnyRef(RetryerFactory.RetryType.FIXED_ATTEMPT.name())) .withValue(RETRY_TIMES, ConfigValueFactory.fromAnyRef(2)) ); TaskRunnerSuiteBase sb = builder.setInstanceName("TestInstance") .setApplicationName("TestApplication") .setAppWorkPath(appWorkDir) .setContainerMetrics(Optional.absent()) .setFileSystem(localFs) .setJobHelixManager(this.helixManager) .setApplicationId("TestApplication-1") .build(); GobblinHelixTaskFactory gobblinHelixTaskFactory = new GobblinHelixTaskFactory(builder, sb.metricContext, this.taskStateTracker, ConfigFactory.empty(), Optional.of(taskDriver)); this.gobblinHelixTask = (GobblinHelixTask) gobblinHelixTaskFactory.createNewTask(taskCallbackContext); TaskRunnerSuiteBase.Builder builderSpy = Mockito.spy(builder); when(builderSpy.getFs()).thenThrow(new RuntimeException("failure on purpose")); gobblinHelixTaskFactory = new GobblinHelixTaskFactory(builderSpy, sb.metricContext, this.taskStateTracker, ConfigFactory.empty(), Optional.of(taskDriver)); try { gobblinHelixTaskFactory.createNewTask(taskCallbackContext).run(); } catch (Throwable t){ return; } Assert.fail(); }
@Override public TaskResult run() { this.taskMetrics.helixTaskTotalRunning.incrementAndGet(); long startTime = System.currentTimeMillis(); log.info("Actual task {} started. [{} {}]", this.taskId, this.applicationName, this.instanceName); try (Closer closer = Closer.create()) { closer.register(MDC.putCloseable(ConfigurationKeys.JOB_NAME_KEY, this.jobName)); closer.register(MDC.putCloseable(ConfigurationKeys.JOB_KEY_KEY, this.jobKey)); this.task.run(); log.info("Actual task {} completed.", this.taskId); this.taskMetrics.helixTaskTotalCompleted.incrementAndGet(); return new TaskResult(TaskResult.Status.COMPLETED, ""); } catch (InterruptedException ie) { log.error("Interrupting task {}", this.taskId); Thread.currentThread().interrupt(); log.error("Actual task {} interrupted.", this.taskId); this.taskMetrics.helixTaskTotalFailed.incrementAndGet(); return new TaskResult(TaskResult.Status.CANCELED, ""); } catch (TaskCreationException te) { eventBus.post(createTaskCreationEvent("Task Execution")); log.error("Actual task {} failed in creation due to {}, will request new container to schedule it", this.taskId, te.getMessage()); this.taskMetrics.helixTaskTotalCancelled.incrementAndGet(); return new TaskResult(TaskResult.Status.FAILED, Throwables.getStackTraceAsString(te)); } catch (Throwable t) { log.error("Actual task {} failed due to {}", this.taskId, t.getMessage()); this.taskMetrics.helixTaskTotalCancelled.incrementAndGet(); return new TaskResult(TaskResult.Status.FAILED, Throwables.getStackTraceAsString(t)); } finally { this.taskMetrics.helixTaskTotalRunning.decrementAndGet(); this.taskMetrics.updateTimeForTaskExecution(startTime); } }
GobblinHelixTask implements Task { @Override public TaskResult run() { this.taskMetrics.helixTaskTotalRunning.incrementAndGet(); long startTime = System.currentTimeMillis(); log.info("Actual task {} started. [{} {}]", this.taskId, this.applicationName, this.instanceName); try (Closer closer = Closer.create()) { closer.register(MDC.putCloseable(ConfigurationKeys.JOB_NAME_KEY, this.jobName)); closer.register(MDC.putCloseable(ConfigurationKeys.JOB_KEY_KEY, this.jobKey)); this.task.run(); log.info("Actual task {} completed.", this.taskId); this.taskMetrics.helixTaskTotalCompleted.incrementAndGet(); return new TaskResult(TaskResult.Status.COMPLETED, ""); } catch (InterruptedException ie) { log.error("Interrupting task {}", this.taskId); Thread.currentThread().interrupt(); log.error("Actual task {} interrupted.", this.taskId); this.taskMetrics.helixTaskTotalFailed.incrementAndGet(); return new TaskResult(TaskResult.Status.CANCELED, ""); } catch (TaskCreationException te) { eventBus.post(createTaskCreationEvent("Task Execution")); log.error("Actual task {} failed in creation due to {}, will request new container to schedule it", this.taskId, te.getMessage()); this.taskMetrics.helixTaskTotalCancelled.incrementAndGet(); return new TaskResult(TaskResult.Status.FAILED, Throwables.getStackTraceAsString(te)); } catch (Throwable t) { log.error("Actual task {} failed due to {}", this.taskId, t.getMessage()); this.taskMetrics.helixTaskTotalCancelled.incrementAndGet(); return new TaskResult(TaskResult.Status.FAILED, Throwables.getStackTraceAsString(t)); } finally { this.taskMetrics.helixTaskTotalRunning.decrementAndGet(); this.taskMetrics.updateTimeForTaskExecution(startTime); } } }
GobblinHelixTask implements Task { @Override public TaskResult run() { this.taskMetrics.helixTaskTotalRunning.incrementAndGet(); long startTime = System.currentTimeMillis(); log.info("Actual task {} started. [{} {}]", this.taskId, this.applicationName, this.instanceName); try (Closer closer = Closer.create()) { closer.register(MDC.putCloseable(ConfigurationKeys.JOB_NAME_KEY, this.jobName)); closer.register(MDC.putCloseable(ConfigurationKeys.JOB_KEY_KEY, this.jobKey)); this.task.run(); log.info("Actual task {} completed.", this.taskId); this.taskMetrics.helixTaskTotalCompleted.incrementAndGet(); return new TaskResult(TaskResult.Status.COMPLETED, ""); } catch (InterruptedException ie) { log.error("Interrupting task {}", this.taskId); Thread.currentThread().interrupt(); log.error("Actual task {} interrupted.", this.taskId); this.taskMetrics.helixTaskTotalFailed.incrementAndGet(); return new TaskResult(TaskResult.Status.CANCELED, ""); } catch (TaskCreationException te) { eventBus.post(createTaskCreationEvent("Task Execution")); log.error("Actual task {} failed in creation due to {}, will request new container to schedule it", this.taskId, te.getMessage()); this.taskMetrics.helixTaskTotalCancelled.incrementAndGet(); return new TaskResult(TaskResult.Status.FAILED, Throwables.getStackTraceAsString(te)); } catch (Throwable t) { log.error("Actual task {} failed due to {}", this.taskId, t.getMessage()); this.taskMetrics.helixTaskTotalCancelled.incrementAndGet(); return new TaskResult(TaskResult.Status.FAILED, Throwables.getStackTraceAsString(t)); } finally { this.taskMetrics.helixTaskTotalRunning.decrementAndGet(); this.taskMetrics.updateTimeForTaskExecution(startTime); } } GobblinHelixTask(TaskRunnerSuiteBase.Builder builder,
                          TaskCallbackContext taskCallbackContext,
                          TaskAttemptBuilder taskAttemptBuilder,
                          StateStores stateStores,
                          GobblinHelixTaskMetrics taskMetrics,
                          TaskDriver taskDriver); }
GobblinHelixTask implements Task { @Override public TaskResult run() { this.taskMetrics.helixTaskTotalRunning.incrementAndGet(); long startTime = System.currentTimeMillis(); log.info("Actual task {} started. [{} {}]", this.taskId, this.applicationName, this.instanceName); try (Closer closer = Closer.create()) { closer.register(MDC.putCloseable(ConfigurationKeys.JOB_NAME_KEY, this.jobName)); closer.register(MDC.putCloseable(ConfigurationKeys.JOB_KEY_KEY, this.jobKey)); this.task.run(); log.info("Actual task {} completed.", this.taskId); this.taskMetrics.helixTaskTotalCompleted.incrementAndGet(); return new TaskResult(TaskResult.Status.COMPLETED, ""); } catch (InterruptedException ie) { log.error("Interrupting task {}", this.taskId); Thread.currentThread().interrupt(); log.error("Actual task {} interrupted.", this.taskId); this.taskMetrics.helixTaskTotalFailed.incrementAndGet(); return new TaskResult(TaskResult.Status.CANCELED, ""); } catch (TaskCreationException te) { eventBus.post(createTaskCreationEvent("Task Execution")); log.error("Actual task {} failed in creation due to {}, will request new container to schedule it", this.taskId, te.getMessage()); this.taskMetrics.helixTaskTotalCancelled.incrementAndGet(); return new TaskResult(TaskResult.Status.FAILED, Throwables.getStackTraceAsString(te)); } catch (Throwable t) { log.error("Actual task {} failed due to {}", this.taskId, t.getMessage()); this.taskMetrics.helixTaskTotalCancelled.incrementAndGet(); return new TaskResult(TaskResult.Status.FAILED, Throwables.getStackTraceAsString(t)); } finally { this.taskMetrics.helixTaskTotalRunning.decrementAndGet(); this.taskMetrics.updateTimeForTaskExecution(startTime); } } GobblinHelixTask(TaskRunnerSuiteBase.Builder builder,
                          TaskCallbackContext taskCallbackContext,
                          TaskAttemptBuilder taskAttemptBuilder,
                          StateStores stateStores,
                          GobblinHelixTaskMetrics taskMetrics,
                          TaskDriver taskDriver); @Override TaskResult run(); @Override void cancel(); }
GobblinHelixTask implements Task { @Override public TaskResult run() { this.taskMetrics.helixTaskTotalRunning.incrementAndGet(); long startTime = System.currentTimeMillis(); log.info("Actual task {} started. [{} {}]", this.taskId, this.applicationName, this.instanceName); try (Closer closer = Closer.create()) { closer.register(MDC.putCloseable(ConfigurationKeys.JOB_NAME_KEY, this.jobName)); closer.register(MDC.putCloseable(ConfigurationKeys.JOB_KEY_KEY, this.jobKey)); this.task.run(); log.info("Actual task {} completed.", this.taskId); this.taskMetrics.helixTaskTotalCompleted.incrementAndGet(); return new TaskResult(TaskResult.Status.COMPLETED, ""); } catch (InterruptedException ie) { log.error("Interrupting task {}", this.taskId); Thread.currentThread().interrupt(); log.error("Actual task {} interrupted.", this.taskId); this.taskMetrics.helixTaskTotalFailed.incrementAndGet(); return new TaskResult(TaskResult.Status.CANCELED, ""); } catch (TaskCreationException te) { eventBus.post(createTaskCreationEvent("Task Execution")); log.error("Actual task {} failed in creation due to {}, will request new container to schedule it", this.taskId, te.getMessage()); this.taskMetrics.helixTaskTotalCancelled.incrementAndGet(); return new TaskResult(TaskResult.Status.FAILED, Throwables.getStackTraceAsString(te)); } catch (Throwable t) { log.error("Actual task {} failed due to {}", this.taskId, t.getMessage()); this.taskMetrics.helixTaskTotalCancelled.incrementAndGet(); return new TaskResult(TaskResult.Status.FAILED, Throwables.getStackTraceAsString(t)); } finally { this.taskMetrics.helixTaskTotalRunning.decrementAndGet(); this.taskMetrics.updateTimeForTaskExecution(startTime); } } GobblinHelixTask(TaskRunnerSuiteBase.Builder builder,
                          TaskCallbackContext taskCallbackContext,
                          TaskAttemptBuilder taskAttemptBuilder,
                          StateStores stateStores,
                          GobblinHelixTaskMetrics taskMetrics,
                          TaskDriver taskDriver); @Override TaskResult run(); @Override void cancel();  }