@Test(dependsOnMethods = "testPutIntoTaskStateStore") public void testCollectOutputTaskStates() throws Exception { this.taskStateCollectorService.runOneIteration(); Assert.assertEquals(this.jobState.getTaskStates().size(), 2); Assert.assertEquals(this.taskStateMap.size(), 2); Assert.assertEquals(this.taskStateMap.get(TASK_ID_0).getJobId(), JOB_ID); Assert.assertEquals(this.taskStateMap.get(TASK_ID_0).getTaskId(), TASK_ID_0); Assert.assertEquals(this.taskStateMap.get(TASK_ID_1).getJobId(), JOB_ID); Assert.assertEquals(this.taskStateMap.get(TASK_ID_1).getTaskId(), TASK_ID_1); }
private void collectOutputTaskStates() throws IOException { List<String> taskStateNames = taskStateStore.getTableNames(outputTaskStateDir.getName(), new Predicate<String>() { @Override public boolean apply(String input) { return input.endsWith(AbstractJobLauncher.TASK_STATE_STORE_TABLE_SUFFIX) && !input.startsWith(FsStateStore.TMP_FILE_PREFIX); }}); if (taskStateNames == null || taskStateNames.size() == 0) { LOGGER.debug("No output task state files found in " + this.outputTaskStateDir); return; } final Queue<TaskState> taskStateQueue = Queues.newConcurrentLinkedQueue(); try (ParallelRunner stateSerDeRunner = new ParallelRunner(this.stateSerDeRunnerThreads, null)) { for (final String taskStateName : taskStateNames) { LOGGER.debug("Found output task state file " + taskStateName); stateSerDeRunner.submitCallable(new Callable<Void>() { @Override public Void call() throws Exception { TaskState taskState = taskStateStore.getAll(outputTaskStateDir.getName(), taskStateName).get(0); taskStateQueue.add(taskState); taskStateStore.delete(outputTaskStateDir.getName(), taskStateName); return null; } }, "Deserialize state for " + taskStateName); } } catch (IOException ioe) { LOGGER.warn("Could not read all task state files."); } LOGGER.info(String.format("Collected task state of %d completed tasks", taskStateQueue.size())); for (TaskState taskState : taskStateQueue) { taskState.setJobState(this.jobState); this.jobState.addTaskState(taskState); } if (optionalTaskCollectorHandler.isPresent()) { LOGGER.info("Execute Pipelined TaskStateCollectorService Handler for " + taskStateQueue.size() + " tasks"); try { optionalTaskCollectorHandler.get().handle(taskStateQueue); } catch (Throwable t) { if (isJobProceedOnCollectorServiceFailure) { log.error("Failed to commit dataset while job proceeds", t); SafeDatasetCommit.setTaskFailureException(taskStateQueue, t); } else { throw new RuntimeException("Hive Registration as the TaskStateCollectorServiceHandler failed.", t); } } } this.eventBus.post(new NewTaskCompletionEvent(ImmutableList.copyOf(taskStateQueue))); }
TaskStateCollectorService extends AbstractScheduledService { private void collectOutputTaskStates() throws IOException { List<String> taskStateNames = taskStateStore.getTableNames(outputTaskStateDir.getName(), new Predicate<String>() { @Override public boolean apply(String input) { return input.endsWith(AbstractJobLauncher.TASK_STATE_STORE_TABLE_SUFFIX) && !input.startsWith(FsStateStore.TMP_FILE_PREFIX); }}); if (taskStateNames == null || taskStateNames.size() == 0) { LOGGER.debug("No output task state files found in " + this.outputTaskStateDir); return; } final Queue<TaskState> taskStateQueue = Queues.newConcurrentLinkedQueue(); try (ParallelRunner stateSerDeRunner = new ParallelRunner(this.stateSerDeRunnerThreads, null)) { for (final String taskStateName : taskStateNames) { LOGGER.debug("Found output task state file " + taskStateName); stateSerDeRunner.submitCallable(new Callable<Void>() { @Override public Void call() throws Exception { TaskState taskState = taskStateStore.getAll(outputTaskStateDir.getName(), taskStateName).get(0); taskStateQueue.add(taskState); taskStateStore.delete(outputTaskStateDir.getName(), taskStateName); return null; } }, "Deserialize state for " + taskStateName); } } catch (IOException ioe) { LOGGER.warn("Could not read all task state files."); } LOGGER.info(String.format("Collected task state of %d completed tasks", taskStateQueue.size())); for (TaskState taskState : taskStateQueue) { taskState.setJobState(this.jobState); this.jobState.addTaskState(taskState); } if (optionalTaskCollectorHandler.isPresent()) { LOGGER.info("Execute Pipelined TaskStateCollectorService Handler for " + taskStateQueue.size() + " tasks"); try { optionalTaskCollectorHandler.get().handle(taskStateQueue); } catch (Throwable t) { if (isJobProceedOnCollectorServiceFailure) { log.error("Failed to commit dataset while job proceeds", t); SafeDatasetCommit.setTaskFailureException(taskStateQueue, t); } else { throw new RuntimeException("Hive Registration as the TaskStateCollectorServiceHandler failed.", t); } } } this.eventBus.post(new NewTaskCompletionEvent(ImmutableList.copyOf(taskStateQueue))); } }
TaskStateCollectorService extends AbstractScheduledService { private void collectOutputTaskStates() throws IOException { List<String> taskStateNames = taskStateStore.getTableNames(outputTaskStateDir.getName(), new Predicate<String>() { @Override public boolean apply(String input) { return input.endsWith(AbstractJobLauncher.TASK_STATE_STORE_TABLE_SUFFIX) && !input.startsWith(FsStateStore.TMP_FILE_PREFIX); }}); if (taskStateNames == null || taskStateNames.size() == 0) { LOGGER.debug("No output task state files found in " + this.outputTaskStateDir); return; } final Queue<TaskState> taskStateQueue = Queues.newConcurrentLinkedQueue(); try (ParallelRunner stateSerDeRunner = new ParallelRunner(this.stateSerDeRunnerThreads, null)) { for (final String taskStateName : taskStateNames) { LOGGER.debug("Found output task state file " + taskStateName); stateSerDeRunner.submitCallable(new Callable<Void>() { @Override public Void call() throws Exception { TaskState taskState = taskStateStore.getAll(outputTaskStateDir.getName(), taskStateName).get(0); taskStateQueue.add(taskState); taskStateStore.delete(outputTaskStateDir.getName(), taskStateName); return null; } }, "Deserialize state for " + taskStateName); } } catch (IOException ioe) { LOGGER.warn("Could not read all task state files."); } LOGGER.info(String.format("Collected task state of %d completed tasks", taskStateQueue.size())); for (TaskState taskState : taskStateQueue) { taskState.setJobState(this.jobState); this.jobState.addTaskState(taskState); } if (optionalTaskCollectorHandler.isPresent()) { LOGGER.info("Execute Pipelined TaskStateCollectorService Handler for " + taskStateQueue.size() + " tasks"); try { optionalTaskCollectorHandler.get().handle(taskStateQueue); } catch (Throwable t) { if (isJobProceedOnCollectorServiceFailure) { log.error("Failed to commit dataset while job proceeds", t); SafeDatasetCommit.setTaskFailureException(taskStateQueue, t); } else { throw new RuntimeException("Hive Registration as the TaskStateCollectorServiceHandler failed.", t); } } } this.eventBus.post(new NewTaskCompletionEvent(ImmutableList.copyOf(taskStateQueue))); } TaskStateCollectorService(Properties jobProps, JobState jobState, EventBus eventBus,
      StateStore<TaskState> taskStateStore, Path outputTaskStateDir); }
TaskStateCollectorService extends AbstractScheduledService { private void collectOutputTaskStates() throws IOException { List<String> taskStateNames = taskStateStore.getTableNames(outputTaskStateDir.getName(), new Predicate<String>() { @Override public boolean apply(String input) { return input.endsWith(AbstractJobLauncher.TASK_STATE_STORE_TABLE_SUFFIX) && !input.startsWith(FsStateStore.TMP_FILE_PREFIX); }}); if (taskStateNames == null || taskStateNames.size() == 0) { LOGGER.debug("No output task state files found in " + this.outputTaskStateDir); return; } final Queue<TaskState> taskStateQueue = Queues.newConcurrentLinkedQueue(); try (ParallelRunner stateSerDeRunner = new ParallelRunner(this.stateSerDeRunnerThreads, null)) { for (final String taskStateName : taskStateNames) { LOGGER.debug("Found output task state file " + taskStateName); stateSerDeRunner.submitCallable(new Callable<Void>() { @Override public Void call() throws Exception { TaskState taskState = taskStateStore.getAll(outputTaskStateDir.getName(), taskStateName).get(0); taskStateQueue.add(taskState); taskStateStore.delete(outputTaskStateDir.getName(), taskStateName); return null; } }, "Deserialize state for " + taskStateName); } } catch (IOException ioe) { LOGGER.warn("Could not read all task state files."); } LOGGER.info(String.format("Collected task state of %d completed tasks", taskStateQueue.size())); for (TaskState taskState : taskStateQueue) { taskState.setJobState(this.jobState); this.jobState.addTaskState(taskState); } if (optionalTaskCollectorHandler.isPresent()) { LOGGER.info("Execute Pipelined TaskStateCollectorService Handler for " + taskStateQueue.size() + " tasks"); try { optionalTaskCollectorHandler.get().handle(taskStateQueue); } catch (Throwable t) { if (isJobProceedOnCollectorServiceFailure) { log.error("Failed to commit dataset while job proceeds", t); SafeDatasetCommit.setTaskFailureException(taskStateQueue, t); } else { throw new RuntimeException("Hive Registration as the TaskStateCollectorServiceHandler failed.", t); } } } this.eventBus.post(new NewTaskCompletionEvent(ImmutableList.copyOf(taskStateQueue))); } TaskStateCollectorService(Properties jobProps, JobState jobState, EventBus eventBus,
      StateStore<TaskState> taskStateStore, Path outputTaskStateDir);  }
TaskStateCollectorService extends AbstractScheduledService { private void collectOutputTaskStates() throws IOException { List<String> taskStateNames = taskStateStore.getTableNames(outputTaskStateDir.getName(), new Predicate<String>() { @Override public boolean apply(String input) { return input.endsWith(AbstractJobLauncher.TASK_STATE_STORE_TABLE_SUFFIX) && !input.startsWith(FsStateStore.TMP_FILE_PREFIX); }}); if (taskStateNames == null || taskStateNames.size() == 0) { LOGGER.debug("No output task state files found in " + this.outputTaskStateDir); return; } final Queue<TaskState> taskStateQueue = Queues.newConcurrentLinkedQueue(); try (ParallelRunner stateSerDeRunner = new ParallelRunner(this.stateSerDeRunnerThreads, null)) { for (final String taskStateName : taskStateNames) { LOGGER.debug("Found output task state file " + taskStateName); stateSerDeRunner.submitCallable(new Callable<Void>() { @Override public Void call() throws Exception { TaskState taskState = taskStateStore.getAll(outputTaskStateDir.getName(), taskStateName).get(0); taskStateQueue.add(taskState); taskStateStore.delete(outputTaskStateDir.getName(), taskStateName); return null; } }, "Deserialize state for " + taskStateName); } } catch (IOException ioe) { LOGGER.warn("Could not read all task state files."); } LOGGER.info(String.format("Collected task state of %d completed tasks", taskStateQueue.size())); for (TaskState taskState : taskStateQueue) { taskState.setJobState(this.jobState); this.jobState.addTaskState(taskState); } if (optionalTaskCollectorHandler.isPresent()) { LOGGER.info("Execute Pipelined TaskStateCollectorService Handler for " + taskStateQueue.size() + " tasks"); try { optionalTaskCollectorHandler.get().handle(taskStateQueue); } catch (Throwable t) { if (isJobProceedOnCollectorServiceFailure) { log.error("Failed to commit dataset while job proceeds", t); SafeDatasetCommit.setTaskFailureException(taskStateQueue, t); } else { throw new RuntimeException("Hive Registration as the TaskStateCollectorServiceHandler failed.", t); } } } this.eventBus.post(new NewTaskCompletionEvent(ImmutableList.copyOf(taskStateQueue))); } TaskStateCollectorService(Properties jobProps, JobState jobState, EventBus eventBus,
      StateStore<TaskState> taskStateStore, Path outputTaskStateDir);   }