@Test public void submitRunTaskCommand() throws Exception { long jobId = 0L; int taskId = 1; JobConfig jobConfig = new TestPlanConfig("/test"); long workerId = 2L; List<Integer> args = Lists.newArrayList(1); mManager.submitRunTaskCommand(jobId, taskId, jobConfig, args, workerId); List<JobCommand> commands = mManager.pollAllPendingCommands(workerId); Assert.assertEquals(1, commands.size()); JobCommand command = commands.get(0); Assert.assertEquals(jobId, command.getRunTaskCommand().getJobId()); Assert.assertEquals(taskId, command.getRunTaskCommand().getTaskId()); Assert.assertEquals(jobConfig, SerializationUtils.deserialize(command.getRunTaskCommand().getJobConfig().toByteArray())); Assert.assertEquals(args, SerializationUtils.deserialize(command.getRunTaskCommand().getTaskArgs().toByteArray())); }
public synchronized void submitRunTaskCommand(long jobId, long taskId, JobConfig jobConfig, Object taskArgs, long workerId) { RunTaskCommand.Builder runTaskCommand = RunTaskCommand.newBuilder(); runTaskCommand.setJobId(jobId); runTaskCommand.setTaskId(taskId); try { runTaskCommand.setJobConfig(ByteString.copyFrom(SerializationUtils.serialize(jobConfig))); if (taskArgs != null) { runTaskCommand.setTaskArgs(ByteString.copyFrom(SerializationUtils.serialize(taskArgs))); } } catch (IOException e) { LOG.info("Failed to serialize the run task command:" + e); return; } JobCommand.Builder command = JobCommand.newBuilder(); command.setRunTaskCommand(runTaskCommand); submit(workerId, command); }
CommandManager { public synchronized void submitRunTaskCommand(long jobId, long taskId, JobConfig jobConfig, Object taskArgs, long workerId) { RunTaskCommand.Builder runTaskCommand = RunTaskCommand.newBuilder(); runTaskCommand.setJobId(jobId); runTaskCommand.setTaskId(taskId); try { runTaskCommand.setJobConfig(ByteString.copyFrom(SerializationUtils.serialize(jobConfig))); if (taskArgs != null) { runTaskCommand.setTaskArgs(ByteString.copyFrom(SerializationUtils.serialize(taskArgs))); } } catch (IOException e) { LOG.info("Failed to serialize the run task command:" + e); return; } JobCommand.Builder command = JobCommand.newBuilder(); command.setRunTaskCommand(runTaskCommand); submit(workerId, command); } }
CommandManager { public synchronized void submitRunTaskCommand(long jobId, long taskId, JobConfig jobConfig, Object taskArgs, long workerId) { RunTaskCommand.Builder runTaskCommand = RunTaskCommand.newBuilder(); runTaskCommand.setJobId(jobId); runTaskCommand.setTaskId(taskId); try { runTaskCommand.setJobConfig(ByteString.copyFrom(SerializationUtils.serialize(jobConfig))); if (taskArgs != null) { runTaskCommand.setTaskArgs(ByteString.copyFrom(SerializationUtils.serialize(taskArgs))); } } catch (IOException e) { LOG.info("Failed to serialize the run task command:" + e); return; } JobCommand.Builder command = JobCommand.newBuilder(); command.setRunTaskCommand(runTaskCommand); submit(workerId, command); } CommandManager(); }
CommandManager { public synchronized void submitRunTaskCommand(long jobId, long taskId, JobConfig jobConfig, Object taskArgs, long workerId) { RunTaskCommand.Builder runTaskCommand = RunTaskCommand.newBuilder(); runTaskCommand.setJobId(jobId); runTaskCommand.setTaskId(taskId); try { runTaskCommand.setJobConfig(ByteString.copyFrom(SerializationUtils.serialize(jobConfig))); if (taskArgs != null) { runTaskCommand.setTaskArgs(ByteString.copyFrom(SerializationUtils.serialize(taskArgs))); } } catch (IOException e) { LOG.info("Failed to serialize the run task command:" + e); return; } JobCommand.Builder command = JobCommand.newBuilder(); command.setRunTaskCommand(runTaskCommand); submit(workerId, command); } CommandManager(); synchronized void submitRunTaskCommand(long jobId, long taskId, JobConfig jobConfig,
      Object taskArgs, long workerId); synchronized void submitCancelTaskCommand(long jobId, long taskId, long workerId); synchronized void submitSetTaskPoolSizeCommand(long workerId, int taskPoolSize); synchronized List<alluxio.grpc.JobCommand> pollAllPendingCommands(long workerId); }
CommandManager { public synchronized void submitRunTaskCommand(long jobId, long taskId, JobConfig jobConfig, Object taskArgs, long workerId) { RunTaskCommand.Builder runTaskCommand = RunTaskCommand.newBuilder(); runTaskCommand.setJobId(jobId); runTaskCommand.setTaskId(taskId); try { runTaskCommand.setJobConfig(ByteString.copyFrom(SerializationUtils.serialize(jobConfig))); if (taskArgs != null) { runTaskCommand.setTaskArgs(ByteString.copyFrom(SerializationUtils.serialize(taskArgs))); } } catch (IOException e) { LOG.info("Failed to serialize the run task command:" + e); return; } JobCommand.Builder command = JobCommand.newBuilder(); command.setRunTaskCommand(runTaskCommand); submit(workerId, command); } CommandManager(); synchronized void submitRunTaskCommand(long jobId, long taskId, JobConfig jobConfig,
      Object taskArgs, long workerId); synchronized void submitCancelTaskCommand(long jobId, long taskId, long workerId); synchronized void submitSetTaskPoolSizeCommand(long workerId, int taskPoolSize); synchronized List<alluxio.grpc.JobCommand> pollAllPendingCommands(long workerId);  }