code: protected Job createJob() throws IOException {
      Job job = Job.getInstance(new Configuration());
      for (Map.Entry<Object, Object> entry : this.taskContext.getTaskState().getProperties().entrySet()) {
        if (entry.getKey() instanceof String && ((String) entry.getKey()).startsWith(JOB_CONFIGURATION_PREFIX)) {
          String actualKey = ((String) entry.getKey()).substring(JOB_CONFIGURATION_PREFIX.length());
          job.getConfiguration().set(actualKey, (String) entry.getValue());
        }
      }
      return job;
    }
docs: Create the {@link Job} to run in this task.
  @return the {@link Job} to run. If this method returns null, no job will be run and the task will be marked as successful.