Read temporary configuration files. If configuration files are not found,
assume an empty job object is being created.

@throws IOException if reading configuration files failed
@throws IllegalStateException if configuration files are missing
485149
private void read() throws IOException {
        lastModified = jobFile.lastModified();
        if (jobFile.exists()) {
            try (final InputStream in = new FileInputStream(jobFile)) {
                final XMLReader parser = XMLUtils.getXMLReader();
                parser.setContentHandler(new JobHandler(prop, files));

                parser.parse(new InputSource(in));
            } catch (final SAXException e) {
                throw new IOException("Failed to read job file: " + e.getMessage());
            }
        } else {
            // defaults
            prop.put(PROPERTY_GENERATE_COPY_OUTER, Generate.NOT_GENERATEOUTTER.toString());
            prop.put(PROPERTY_ONLY_TOPIC_IN_MAP, Boolean.toString(false));
            prop.put(PROPERTY_OUTER_CONTROL, OutterControl.WARN.toString());
        }
    }
1008857
public void read() throws IOException {
		long headerSize = header.getHeaderSize();
		int loadModuleSize = getLoadModuleSize();

		try (RandomAccessFile raf = new RandomAccessFile(file, "r")) {
			raf.seek(headerSize);
			loadModuleBytes = new byte[loadModuleSize];
			raf.readFully(loadModuleBytes);
		}
	}
1281251
public Set<String> configureJob(Job job) throws FileNotFoundException, IOException {
    Set<String> instanceFiles = new HashSet<String>();
    for (Map.Entry<Path, List<Input>> entry : multiInputs.entrySet()) {
      for (int inputId = 0; inputId < entry.getValue().size(); inputId++) {
        Input input = entry.getValue().get(inputId);
        instanceFiles.addAll(PangoolMultipleInputs.addInputPath(job, input.path, input.inputFormat,
            input.inputProcessor, input.specificContext, inputId));
      }
    }
    return instanceFiles;
  }
1329472
private void copyAndConfigureFiles(JobConf job, Path uploadFileDir,
      boolean shared)
    throws IOException {

    if (!(job.getBoolean("mapred.used.genericoptionsparser", false))) {
      LOG.warn("Use GenericOptionsParser for parsing the arguments. " +
               "Applications should implement Tool for the same.");
    }

    // get all the command line arguments into the
    // jobconf passed in by the user conf
    String files = job.get("tmpfiles");
    String libjars = job.get("tmpjars");
    String archives = job.get("tmparchives");

    //
    // Figure out what fs the JobTracker is using.  Copy the
    // job to it, under a temporary name.  This allows DFS to work,
    // and under the local fs also provides UNIX-like object loading
    // semantics.  (that is, if the job file is deleted right after
    // submission, we can still run the submission to completion)
    //

    // Create a number of filenames in the JobTracker's fs namespace
    FileSystem fs = getFs();
    LOG.debug("default FileSystem: " + fs.getUri());

    // We know file system of distributed cache, initialize index
    if (filesInCache == null) {
      filesInCache = new DistributedCacheIndex(fs);
    }

    uploadFileDir = fs.makeQualified(uploadFileDir);
    uploadFileDir = new Path(uploadFileDir.toUri().getPath());
    FsPermission mapredSysPerms = new FsPermission(JOB_DIR_PERMISSION);

    if (!fs.exists(uploadFileDir)) {
      FileSystem.mkdirs(fs, uploadFileDir, mapredSysPerms);
    }
    Path filesDir = new Path(uploadFileDir, "files");
    Path archivesDir = new Path(uploadFileDir, "archives");
    Path libjarsDir = new Path(uploadFileDir, "libjars");
    short replication = (short)job.getInt("mapred.submit.replication", 10);

    fileInfo = new HashMap<URI, FileInfo>();

    String originalJar = job.getJar();

    if (originalJar != null) {
      // use jar name if job is not named.
      if ("".equals(job.getJobName())) {
        job.setJobName(new Path(originalJar).getName());
      }

      Path uploadJarPath;

      Path originalJarPath = new Path(originalJar);
      originalJarPath =
        originalJarPath.makeQualified(FileSystem.getLocal(job));

      try {
        // If sharing is turned on, we load the job jar into the distributed
        // cache

        if (shared) {
          if (!fs.exists(libjarsDir)) {
            FileSystem.mkdirs(fs, libjarsDir, mapredSysPerms);
          }

          MD5Hash md5hash = MD5Hash.digest(new
              FileInputStream(originalJarPath.toUri().getPath()));
          uploadJarPath = copyRemoteFiles(fs, libjarsDir, originalJarPath, job,
              replication, md5hash.toString());
          URI pathURI = new URI(uploadJarPath.toUri().toString());

          DistributedCache.addSharedArchiveToClassPath(uploadJarPath, job);
          fileInfo.put(pathURI, new FileInfo(md5hash.toString(),
                md5hash.getFileLength(), 0));
        } else {
          // Otherwise we copy jar to JT's filesystem
          uploadJarPath = new Path(uploadFileDir, "job.jar");
          fs.copyFromLocalFile(originalJarPath, uploadJarPath);
        }
      } catch (URISyntaxException ue) {
        // should not throw an uri exception
        throw new IOException("Failed to create uri for " + originalJar);
      }

      job.setJar(uploadJarPath.toString());
      fs.setReplication(uploadJarPath, replication);

      try {
        fs.setPermission(uploadJarPath, new FsPermission(JOB_FILE_PERMISSION));
      } catch (IOException ioe) {
        LOG.warn("Unable to set job jar permission");
      }
    } else {
      LOG.warn("No job jar file set. User classes may not be found. "+
          "See JobConf(Class) or JobConf#setJar(String).");
    }

    // add all the command line files/ jars and archive
    // first copy them to jobtrackers filesystem

    if (files != null) {
      if (!fs.exists(filesDir)) {
        FileSystem.mkdirs(fs, filesDir, mapredSysPerms);
      }

      String[] fileArr = files.split(",");
      for (String tmpFile: fileArr) {
        Path tmp = new Path(tmpFile);

        Path newPath;
        FileStatus fStatus = null;
        MD5Hash md5hash = null;
        try {
          if (shared) {
            md5hash
              = MD5Hash.digest(new FileInputStream(tmp.toUri().getPath()));
            newPath = copyRemoteFiles(fs, filesDir, tmp, job, replication,
                md5hash.toString());

            URI pathURI = new URI(newPath.toUri().toString() + "#" +
                newPath.getName());

            DistributedCache.addSharedCacheFile(pathURI, job);
            fileInfo.put(pathURI, new FileInfo(md5hash.toString(),
                  md5hash.getFileLength(),
                  0));

          } else {
            newPath = copyRemoteFiles(fs, filesDir, tmp, job, replication);
            fStatus = DistributedCache.getFileStatus(job, newPath.toUri());

            URI pathURI = new URI(newPath.toUri().toString() + "#" +
                newPath.getName());

            DistributedCache.addCacheFile(pathURI, job);
            fileInfo.put(pathURI, new FileInfo(null,
                  fStatus.getLen(),
                  fStatus.getModificationTime()));
          }

        } catch(URISyntaxException ue) {
          //should not throw a uri exception
          throw new IOException("Failed to create uri for " + tmpFile);
        }
        DistributedCache.createSymlink(job);
      }
    }

    if (libjars != null) {
      if (!fs.exists(libjarsDir)) {
        FileSystem.mkdirs(fs, libjarsDir, mapredSysPerms);
      }

      String[] libjarsArr = libjars.split(",");
      for (String tmpjars: libjarsArr) {
        Path tmp = new Path(tmpjars);

        Path newPath;
        if (shared) {
          MD5Hash md5hash
            = MD5Hash.digest(new FileInputStream(tmp.toUri().getPath()));
          newPath = copyRemoteFiles(fs, libjarsDir, tmp, job, replication,
              md5hash.toString());
          DistributedCache.addSharedArchiveToClassPath(newPath, job);

          fileInfo.put(newPath.makeQualified(newPath.getFileSystem(job)).toUri(),
                       new FileInfo(md5hash.toString(),
                                    md5hash.getFileLength(),
                                    0));
        } else {
          newPath = copyRemoteFiles(fs, libjarsDir, tmp, job, replication);
          DistributedCache.addArchiveToClassPath(newPath, job);

          FileStatus fStatus = DistributedCache.getFileStatus(job,
              newPath.toUri());
          fileInfo.put(newPath.makeQualified(newPath.getFileSystem(job)).toUri(),
                       new FileInfo(null,
                                    fStatus.getLen(),
                                    fStatus.getModificationTime()));
        }
      }
    }

    if (archives != null) {
      if (!fs.exists(archivesDir)) {
        FileSystem.mkdirs(fs, archivesDir, mapredSysPerms);
      }

     String[] archivesArr = archives.split(",");
     for (String tmpArchives: archivesArr) {
       Path tmp = new Path(tmpArchives);

       Path newPath;
       MD5Hash md5hash = null;
       FileStatus fStatus = null;
       try {
         if (shared) {
           md5hash
             = MD5Hash.digest(new FileInputStream(tmp.toUri().getPath()));
           newPath = copyRemoteFiles(fs, archivesDir, tmp, job, replication,
               md5hash.toString());
           URI pathURI = new URI(newPath.toUri().toString() + "#" +
               newPath.getName());

           DistributedCache.addSharedCacheArchive(pathURI, job);
           fileInfo.put(pathURI, new FileInfo(md5hash.toString(),
                 md5hash.getFileLength(),
                 0));

         } else {
           newPath = copyRemoteFiles(fs, archivesDir, tmp, job, replication);
           fStatus = DistributedCache.getFileStatus(job, newPath.toUri());

           URI pathURI = new URI(newPath.toUri().toString() + "#" +
               newPath.getName());

           DistributedCache.addCacheArchive(pathURI, job);
           fileInfo.put(pathURI, new FileInfo(null,
                 fStatus.getLen(),
                 fStatus.getModificationTime()));
         }
       } catch(URISyntaxException ue) {
         //should not throw an uri excpetion
         throw new IOException("Failed to create uri for " + tmpArchives);
       }
       DistributedCache.createSymlink(job);
     }
    }

    // set the timestamps and md5 of the archives and files
    URI[] tarchives = DistributedCache.getSharedCacheArchives(job);
    if (tarchives != null) {
      StringBuffer archiveLength = new StringBuffer();
      FileStatus fStatus;
      FileInfo info;
      long fileLength;

      for (int i = 0; i < tarchives.length; i++) {
        if (i != 0) archiveLength.append(',');

        info = fileInfo.get(tarchives[i]);
        if (info == null) {
          fStatus = DistributedCache.getFileStatus(job, tarchives[i]);
          fileLength = fStatus.getLen();
        } else {
          fileLength = info.fileLength;
        }

        archiveLength.append(fileLength);
      }

      DistributedCache.setSharedArchiveLength(job, archiveLength.toString());
    }

    URI[] tfiles = DistributedCache.getSharedCacheFiles(job);
    if (tfiles != null) {
      StringBuffer fileLength = new StringBuffer();
      FileStatus fStatus;
      FileInfo info;
      long len;

      for (int i = 0; i < tfiles.length; i++) {
        if (i != 0) fileLength.append(',');

        info = fileInfo.get(tfiles[i]);
        if (info == null) {
          fStatus = DistributedCache.getFileStatus(job, tfiles[i]);
          len = fStatus.getLen();
        } else {
          len = info.fileLength;
        }

        fileLength.append(len);
      }

      DistributedCache.setSharedFileLength(job, fileLength.toString());
    }

    tarchives = DistributedCache.getCacheArchives(job);
    if (tarchives != null) {
      StringBuffer archiveTimestamps = new StringBuffer();
      FileInfo info;
      long timeStamp;

      for (int i = 0; i < tarchives.length; i++) {
        if (i != 0) archiveTimestamps.append(',');

        info = fileInfo.get(tarchives[i]);
        if (info == null) {
          timeStamp = DistributedCache.getTimestamp(job, tarchives[i]);
        } else {
          timeStamp = info.timeStamp;
        }
        archiveTimestamps.append(timeStamp);
      }

      DistributedCache.setArchiveTimestamps(job, archiveTimestamps.toString());
    }

    tfiles = DistributedCache.getCacheFiles(job);
    if (tfiles != null) {
      StringBuffer fileTimestamps = new StringBuffer();
      FileInfo info;
      long timeStamp;

      for (int i = 0; i < tfiles.length; i++) {
        if (i != 0) fileTimestamps.append(',');

        info = fileInfo.get(tfiles[i]);
        if (info == null) {
          timeStamp = DistributedCache.getTimestamp(job, tfiles[i]);
        } else {
          timeStamp = info.timeStamp;
        }
        fileTimestamps.append(timeStamp);
      }

      DistributedCache.setFileTimestamps(job, fileTimestamps.toString());
    }

    configureUserName(job);

  }
6849
private byte[] readJobFile(FileStatus fileStatus) throws IOException {
    byte[] rawBytes = null;
    FSDataInputStream fsdis = null;
    try {
      long fileLength = fileStatus.getLen();
      int fileLengthInt = (int) fileLength;
      rawBytes = new byte[fileLengthInt];
      fsdis = hdfs.open(fileStatus.getPath());
      IOUtils.readFully(fsdis, rawBytes, 0, fileLengthInt);
    } finally {
      IOUtils.closeStream(fsdis);
    }
    return rawBytes;
  }
966374
@Override
	public void readFrom(InputStream is) throws IOException {
		this.currentItem = 0;
		// read from source and write into tempfile
		// open temp file depending on options compressed/encrypted
		OutputStream tempOS = null;
		if ((this.ca != null) || (this.compressTempFile)) {
			tempOS = new FileOutputStream(this.tempFile);
			if (this.ca != null) { // encrypt file if configured
				tempOS = new CipherOutputStream(tempOS, this.ciEncrypt);
			}
			if (this.compressTempFile) { // compress file if configured
				tempOS = new GZIPOutputStream(tempOS, EncryptedCachedDiskStringsTable.compressBufferSize);
			}
		} else { // not encrypted and not compressed: configure a random access file for
					// writing/reading = highest performance
			this.tempRAF = new RandomAccessFile(this.tempFile, "rw");
		}
		// read from source
		// use Stax event reader
		XMLEventReader xer = null;
		try {
			xer = StaxHelper.newXMLInputFactory().createXMLEventReader(this.originalIS);
			while (xer.hasNext()) {
				XMLEvent xe = xer.nextEvent();
				// check if it is a string item (entry in string table)
				if (xe.isStartElement() && xe.asStartElement().getName().getLocalPart().equalsIgnoreCase("si")) {
					String siText = this.parseSIText(xer);
					// add string to temp file
					this.addString(siText, tempOS);
					this.count++;
				}
			}
			// close tempfile
			// make tempFile available as a reader
		} catch (XMLStreamException e) {
			LOG.error("Cannot read original SharedStringTable from document. Exception " + e);
			throw new IOException(e);
		} catch (FormatNotUnderstoodException e) {
			LOG.error("Cannot read properly SharedStringTable from document. Exception " + e);
			throw new IOException(e);
		} finally {
			// close temporary Stream (tempfile should be deleted using the close method of
			// this class and not here)
			if (tempOS != null) {
				tempOS.close();
			}
		}
		// open the input stream towards the temp file
		this.accessTempFile(0L);

	}
1235704
@Benchmark
	@BenchmarkMode(Mode.AverageTime)
	public void byteBufferFileOutputStream(final Configuration configuration) throws IOException {
		ByteBuffer buffer = ByteBuffer.allocate(BUFFER_CAPACITY);

		try (OutputStream stream = new FileOutputStream(configuration.file)) {
			for (long i = 0; i < LINES; ++i) {
				if (buffer.remaining() < DATA.length) {
					stream.write(buffer.array(), 0, buffer.position());
					buffer.rewind();
				}

				if (buffer.remaining() < DATA.length) {
					stream.write(DATA);
				} else {
					buffer.put(DATA);
				}
			}

			if (buffer.position() > 0) {
				stream.write(buffer.array(), 0, buffer.position());
			}
		}
	}
649647
public synchronized File getTempDirectory() throws IOException {
        if (tempDirectory == null) {
            final File baseTemp = new File(getString(Settings.KEYS.TEMP_DIRECTORY, System.getProperty("java.io.tmpdir")));
            tempDirectory = FileUtils.createTempDirectory(baseTemp);
        }
        return tempDirectory;
    }
1036896
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;
  }
193948
public void delete() throws IOException {
		synchronized(tempFiles) {
			for(WeakReference<TempFile> tempFileRef : tempFiles) {
				TempFile tempFile = tempFileRef.get();
				if(tempFile!=null) tempFile.delete();
			}
			tempFiles.clear();
		}
	}
1003468
@SuppressWarnings({"unchecked"})
    public <T> T read(InputStream jsonInputStream) throws IOException {
        return read(jsonInputStream, Configuration.defaultConfiguration());
    }
114991
private void readConfigFile() {

        m_isValid = false;
        if ((null != m_configFile) && m_configFile.exists()) {
            if (m_configFile.canRead()) {
                Properties props = new Properties();
                try (FileInputStream input = new FileInputStream(m_configFile)) {
                    props.load(input);
                    for (Entry<Object, Object> entry : props.entrySet()) {
                        String key = (String)entry.getKey();
                        String propValue = (String)entry.getValue();
                        if (key.equals(DEFAULT_MODULES_TO_EXPORT)) {
                            String value = getValueFromProp(propValue).trim();
                            if (value.contains(",")) {
                                m_configuredModules = Arrays.asList(value.split(" *, *"));
                            } else {
                                m_configuredModules = Arrays.asList(value.split(" +"));
                            }
                        }

                        if (key.equals(DEFAULT_MODULE_PATH)) {
                            m_modulesPath = getValueFromProp(propValue).trim();
                        }

                        if (key.equals(DEFAULT_MODULE_RESOURCES_SUBFOLDER)) {
                            m_resourcesSubfolder = getValueFromProp(propValue).trim();
                        }

                        if (key.equals(DEFAULT_PULL_MODE_BEFORE)) {
                            String value = getValueFromProp(propValue);
                            value = value.trim();
                            if (value.equals("1")) {
                                m_defaultPullModeBefore = true;
                            }
                        }
                        if (key.equals(DEFAULT_PULL_MODE_AFTER)) {
                            String value = getValueFromProp(propValue);
                            value = value.trim();
                            if (value.equals("1")) {
                                m_defaultPullModeAfter = true;
                            }
                        }
                        if (key.equals(DEFAULT_PUSH_MODE)) {
                            String value = getValueFromProp(propValue);
                            value = value.trim();
                            if (value.equals("1")) {
                                m_defaultPushMode = true;
                            }
                        }
                        if (key.equals(DEFAULT_MODULE_EXPORT_PATH)) {
                            String value = getValueFromProp(propValue);
                            m_defaultModuleExportPath = value.trim();
                        }
                        if (key.equals(DEFAULT_REPOSITORY_PATH)) {
                            String value = getValueFromProp(propValue);
                            m_repositoryPath = value.trim();
                        }
                        if (key.equals(DEFAULT_EXPORT_MODE)) {
                            String value = getValueFromProp(propValue);
                            m_defaultExportMode = (value.trim().equals("1")) ? 1 : 0;
                        }
                        if (key.equals(DEFAULT_COMMIT_MESSAGE)) {
                            String value = getValueFromProp(propValue);
                            m_defaultCommitMessage = value;
                        }
                        if (key.equals(DEFAULT_GIT_USER_NAME)) {
                            String value = getValueFromProp(propValue);
                            m_defaultGitUserName = value;
                        }
                        if (key.equals(DEFAULT_GIT_USER_EMAIL)) {
                            String value = getValueFromProp(propValue);
                            m_defaultGitUserEmail = value;
                        }
                        if (key.equals(DEFAULT_EXCLUDE_LIBS)) {
                            String value = getValueFromProp(propValue);
                            m_defaultExcludeLibs = (value.trim().equals("1")) ? true : false;
                        }
                        if (key.equals(DEFAULT_COMMIT_MODE)) {
                            String value = getValueFromProp(propValue);
                            m_defaultCommitMode = (value.trim().equals("1")) ? true : false;
                        }
                        if (key.equals(DEFAULT_IGNORE_UNCLEAN)) {
                            String value = getValueFromProp(propValue);
                            m_defaultIgnoreUnclean = (value.trim().equals("1")) ? true : false;
                        }
                        if (key.equals(DEFAULT_COPY_AND_UNZIP)) {
                            String value = getValueFromProp(propValue);
                            m_defaultCopyAndUnzip = (value.trim().equals("1")) ? true : false;
                        }
                    }
                    m_isValid = true;
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
976474
public File getFile() throws IllegalStateException {
		File f = tempFile;
		if(f==null) throw new IllegalStateException();
		return f;
	}
966373
public FsSettings readAsJson(String jobname) throws IOException {
        return FsSettingsParser.fromJson(readFile(jobname, FILENAME_JSON));
    }
215022
public FsJob read(String jobname) throws IOException {
        return FsJobParser.fromJson(readFile(jobname, FILENAME));
    }
710627
public synchronized void storeConfiguration() throws IOException
    {
        OutputStream out = new FileOutputStream(configFilePath);
        configuration.store(out, configFilePath);
    }
797835
private JobConfig.Builder createJob(List<WorkUnit> workUnits) throws IOException {
    Map<String, TaskConfig> taskConfigMap = Maps.newHashMap();

    try (ParallelRunner stateSerDeRunner = new ParallelRunner(this.stateSerDeRunnerThreads, this.fs)) {
      int multiTaskIdSequence = 0;
      for (WorkUnit workUnit : workUnits) {
        if (workUnit instanceof MultiWorkUnit) {
          workUnit.setId(JobLauncherUtils.newMultiTaskId(this.jobContext.getJobId(), multiTaskIdSequence++));
        }
        addWorkUnit(workUnit, stateSerDeRunner, taskConfigMap);
      }

      Path jobStateFilePath;

      // write the job.state using the state store if present, otherwise serialize directly to the file
      if (this.stateStores.haveJobStateStore()) {
        jobStateFilePath = GobblinClusterUtils.getJobStateFilePath(true, this.appWorkDir, this.jobContext.getJobId());
        this.stateStores.getJobStateStore().put(jobStateFilePath.getParent().getName(), jobStateFilePath.getName(),
            this.jobContext.getJobState());
      } else {
        jobStateFilePath = GobblinClusterUtils.getJobStateFilePath(false, this.appWorkDir, this.jobContext.getJobId());
        SerializationUtils.serializeState(this.fs, jobStateFilePath, this.jobContext.getJobState());
      }

      LOGGER.debug("GobblinHelixJobLauncher.createJob: jobStateFilePath {}, jobState {} jobProperties {}",
          jobStateFilePath, this.jobContext.getJobState().toString(), this.jobContext.getJobState().getProperties());
    }

    JobConfig.Builder jobConfigBuilder = new JobConfig.Builder();

    // Helix task attempts = retries + 1 (fallback to general task retry for backward compatibility)
    jobConfigBuilder.setMaxAttemptsPerTask(this.jobContext.getJobState().getPropAsInt(
        GobblinClusterConfigurationKeys.HELIX_TASK_MAX_ATTEMPTS_KEY,
        this.jobContext.getJobState().getPropAsInt(
            ConfigurationKeys.MAX_TASK_RETRIES_KEY,
            ConfigurationKeys.DEFAULT_MAX_TASK_RETRIES)) + 1);

    // Helix task timeout (fallback to general task timeout for backward compatibility)
    jobConfigBuilder.setTimeoutPerTask(this.jobContext.getJobState().getPropAsLong(
        GobblinClusterConfigurationKeys.HELIX_TASK_TIMEOUT_SECONDS,
        this.jobContext.getJobState().getPropAsLong(
            ConfigurationKeys.TASK_TIMEOUT_SECONDS,
            ConfigurationKeys.DEFAULT_TASK_TIMEOUT_SECONDS)) * 1000);

    jobConfigBuilder.setFailureThreshold(workUnits.size());
    jobConfigBuilder.addTaskConfigMap(taskConfigMap).setCommand(GobblinTaskRunner.GOBBLIN_TASK_FACTORY_NAME);
    jobConfigBuilder.setNumConcurrentTasksPerInstance(ConfigUtils.getInt(jobConfig,
        GobblinClusterConfigurationKeys.HELIX_CLUSTER_TASK_CONCURRENCY,
        GobblinClusterConfigurationKeys.HELIX_CLUSTER_TASK_CONCURRENCY_DEFAULT));

    if (this.jobConfig.hasPath(GobblinClusterConfigurationKeys.HELIX_JOB_TAG_KEY)) {
      String jobTag = this.jobConfig.getString(GobblinClusterConfigurationKeys.HELIX_JOB_TAG_KEY);
      log.info("Job {} has tags associated : {}", this.jobContext.getJobId(), jobTag);
      jobConfigBuilder.setInstanceGroupTag(jobTag);
    }

    if (this.jobConfig.hasPath(GobblinClusterConfigurationKeys.HELIX_JOB_TYPE_KEY)) {
      String jobType = this.jobConfig.getString(this.jobConfig.getString(GobblinClusterConfigurationKeys.HELIX_JOB_TYPE_KEY));
      log.info("Job {} has types associated : {}", this.jobContext.getJobId(), jobType);
      jobConfigBuilder.setJobType(jobType);
    }

    if (Task.getExecutionModel(ConfigUtils.configToState(jobConfig)).equals(ExecutionModel.STREAMING)) {
      jobConfigBuilder.setRebalanceRunningTask(true);
    }

    jobConfigBuilder.setExpiry(this.jobContext.getJobState().getPropAsLong(
        GobblinClusterConfigurationKeys.HELIX_WORKFLOW_EXPIRY_TIME_SECONDS, GobblinClusterConfigurationKeys.DEFAULT_HELIX_WORKFLOW_EXPIRY_TIME_SECONDS));

    return jobConfigBuilder;
  }