code: public synchronized static void setJobSpecificOutputPaths(State state) {
  
      // Other tasks may have set this already
      if (!StringUtils.containsIgnoreCase(state.getProp(ConfigurationKeys.WRITER_STAGING_DIR),
          state.getProp(ConfigurationKeys.JOB_ID_KEY))) {
  
        state.setProp(ConfigurationKeys.WRITER_STAGING_DIR, new Path(state.getProp(ConfigurationKeys.WRITER_STAGING_DIR),
            state.getProp(ConfigurationKeys.JOB_ID_KEY)));
        state.setProp(ConfigurationKeys.WRITER_OUTPUT_DIR, new Path(state.getProp(ConfigurationKeys.WRITER_OUTPUT_DIR),
            state.getProp(ConfigurationKeys.JOB_ID_KEY)));
  
      }
  
    }
docs: Each job gets its own task-staging and task-output directory. Update the staging and output directories to
  contain job_id. This is to make sure uncleaned data from previous execution does not corrupt final published data
  produced by this execution.