code: public synchronized void executeTask(long jobId, int taskId, JobConfig jobConfig,
        Serializable taskArgs, RunTaskContext context) {
      Future<?> future = mTaskExecutionService
          .submit(new TaskExecutor(jobId, taskId, jobConfig, taskArgs, context, this));
      Pair<Long, Integer> id = new Pair<>(jobId, taskId);
      mTaskFutures.put(id, future);
      TaskInfo.Builder taskInfo = TaskInfo.newBuilder();
      taskInfo.setJobId(jobId);
      taskInfo.setTaskId(taskId);
      taskInfo.setStatus(Status.RUNNING);
      mUnfinishedTasks.put(id, taskInfo);
      mTaskUpdates.put(id, taskInfo.build());
      LOG.info("Task {} for job {} started", taskId, jobId);
    }
docs: Executes the given task.
  
  @param jobId the job id
  @param taskId the task id
  @param jobConfig the job configuration
  @param taskArgs the arguments
  @param context the context of the worker