@Test public void runCompletion() throws Exception { long jobId = 1; long taskId = 2; JobConfig jobConfig = mock(JobConfig.class); Serializable taskArgs = Lists.newArrayList(1); RunTaskContext context = mock(RunTaskContext.class); Integer taskResult = 1; @SuppressWarnings("unchecked") PlanDefinition<JobConfig, Serializable, Serializable> planDefinition = mock(PlanDefinition.class); when(mRegistry.getJobDefinition(any(JobConfig.class))).thenReturn(planDefinition); when(planDefinition.runTask(any(JobConfig.class), eq(taskArgs), any(RunTaskContext.class))) .thenReturn(taskResult); RunTaskCommand command = RunTaskCommand.newBuilder() .setJobConfig(ByteString.copyFrom(SerializationUtils.serialize(jobConfig))) .setTaskArgs(ByteString.copyFrom(SerializationUtils.serialize(taskArgs))).build(); TaskExecutor executor = new TaskExecutor(jobId, taskId, command, context, mTaskExecutorManager); executor.run(); verify(planDefinition).runTask(any(JobConfig.class), eq(taskArgs), eq(context)); verify(mTaskExecutorManager).notifyTaskCompletion(jobId, taskId, taskResult); }
@Override public void run() { JobConfig jobConfig = null; Serializable taskArgs = null; try { jobConfig = (JobConfig) SerializationUtils.deserialize( mRunTaskCommand.getJobConfig().toByteArray()); if (mRunTaskCommand.hasTaskArgs()) { taskArgs = SerializationUtils.deserialize(mRunTaskCommand.getTaskArgs().toByteArray()); } } catch (IOException | ClassNotFoundException e) { fail(e, jobConfig, null); } PlanDefinition<JobConfig, Serializable, Serializable> definition; try { definition = PlanDefinitionRegistry.INSTANCE.getJobDefinition(jobConfig); } catch (JobDoesNotExistException e) { LOG.error("The job definition for config {} does not exist.", jobConfig.getName()); fail(e, jobConfig, taskArgs); return; } mTaskExecutorManager.notifyTaskRunning(mJobId, mTaskId); Serializable result; try { result = definition.runTask(jobConfig, taskArgs, mContext); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } catch (Throwable t) { fail(t, jobConfig, taskArgs); return; } mTaskExecutorManager.notifyTaskCompletion(mJobId, mTaskId, result); }
TaskExecutor implements Runnable { @Override public void run() { JobConfig jobConfig = null; Serializable taskArgs = null; try { jobConfig = (JobConfig) SerializationUtils.deserialize( mRunTaskCommand.getJobConfig().toByteArray()); if (mRunTaskCommand.hasTaskArgs()) { taskArgs = SerializationUtils.deserialize(mRunTaskCommand.getTaskArgs().toByteArray()); } } catch (IOException | ClassNotFoundException e) { fail(e, jobConfig, null); } PlanDefinition<JobConfig, Serializable, Serializable> definition; try { definition = PlanDefinitionRegistry.INSTANCE.getJobDefinition(jobConfig); } catch (JobDoesNotExistException e) { LOG.error("The job definition for config {} does not exist.", jobConfig.getName()); fail(e, jobConfig, taskArgs); return; } mTaskExecutorManager.notifyTaskRunning(mJobId, mTaskId); Serializable result; try { result = definition.runTask(jobConfig, taskArgs, mContext); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } catch (Throwable t) { fail(t, jobConfig, taskArgs); return; } mTaskExecutorManager.notifyTaskCompletion(mJobId, mTaskId, result); } }
TaskExecutor implements Runnable { @Override public void run() { JobConfig jobConfig = null; Serializable taskArgs = null; try { jobConfig = (JobConfig) SerializationUtils.deserialize( mRunTaskCommand.getJobConfig().toByteArray()); if (mRunTaskCommand.hasTaskArgs()) { taskArgs = SerializationUtils.deserialize(mRunTaskCommand.getTaskArgs().toByteArray()); } } catch (IOException | ClassNotFoundException e) { fail(e, jobConfig, null); } PlanDefinition<JobConfig, Serializable, Serializable> definition; try { definition = PlanDefinitionRegistry.INSTANCE.getJobDefinition(jobConfig); } catch (JobDoesNotExistException e) { LOG.error("The job definition for config {} does not exist.", jobConfig.getName()); fail(e, jobConfig, taskArgs); return; } mTaskExecutorManager.notifyTaskRunning(mJobId, mTaskId); Serializable result; try { result = definition.runTask(jobConfig, taskArgs, mContext); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } catch (Throwable t) { fail(t, jobConfig, taskArgs); return; } mTaskExecutorManager.notifyTaskCompletion(mJobId, mTaskId, result); } TaskExecutor(long jobId, long taskId, RunTaskCommand runTaskCommand,
      RunTaskContext context, TaskExecutorManager taskExecutorManager); }
TaskExecutor implements Runnable { @Override public void run() { JobConfig jobConfig = null; Serializable taskArgs = null; try { jobConfig = (JobConfig) SerializationUtils.deserialize( mRunTaskCommand.getJobConfig().toByteArray()); if (mRunTaskCommand.hasTaskArgs()) { taskArgs = SerializationUtils.deserialize(mRunTaskCommand.getTaskArgs().toByteArray()); } } catch (IOException | ClassNotFoundException e) { fail(e, jobConfig, null); } PlanDefinition<JobConfig, Serializable, Serializable> definition; try { definition = PlanDefinitionRegistry.INSTANCE.getJobDefinition(jobConfig); } catch (JobDoesNotExistException e) { LOG.error("The job definition for config {} does not exist.", jobConfig.getName()); fail(e, jobConfig, taskArgs); return; } mTaskExecutorManager.notifyTaskRunning(mJobId, mTaskId); Serializable result; try { result = definition.runTask(jobConfig, taskArgs, mContext); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } catch (Throwable t) { fail(t, jobConfig, taskArgs); return; } mTaskExecutorManager.notifyTaskCompletion(mJobId, mTaskId, result); } TaskExecutor(long jobId, long taskId, RunTaskCommand runTaskCommand,
      RunTaskContext context, TaskExecutorManager taskExecutorManager); @Override void run(); }
TaskExecutor implements Runnable { @Override public void run() { JobConfig jobConfig = null; Serializable taskArgs = null; try { jobConfig = (JobConfig) SerializationUtils.deserialize( mRunTaskCommand.getJobConfig().toByteArray()); if (mRunTaskCommand.hasTaskArgs()) { taskArgs = SerializationUtils.deserialize(mRunTaskCommand.getTaskArgs().toByteArray()); } } catch (IOException | ClassNotFoundException e) { fail(e, jobConfig, null); } PlanDefinition<JobConfig, Serializable, Serializable> definition; try { definition = PlanDefinitionRegistry.INSTANCE.getJobDefinition(jobConfig); } catch (JobDoesNotExistException e) { LOG.error("The job definition for config {} does not exist.", jobConfig.getName()); fail(e, jobConfig, taskArgs); return; } mTaskExecutorManager.notifyTaskRunning(mJobId, mTaskId); Serializable result; try { result = definition.runTask(jobConfig, taskArgs, mContext); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } catch (Throwable t) { fail(t, jobConfig, taskArgs); return; } mTaskExecutorManager.notifyTaskCompletion(mJobId, mTaskId, result); } TaskExecutor(long jobId, long taskId, RunTaskCommand runTaskCommand,
      RunTaskContext context, TaskExecutorManager taskExecutorManager); @Override void run();  }