@Override
        public Void run() {
          HDFSContext.load();
          loadGeneralConfiguration();
          loadAnalyticJobGenerator();
          ElephantContext.init();

          // Initialize the metrics registries.
          MetricsController.init();

          logger.info("executor num is " + _executorNum);
          if (_executorNum < 1) {
            throw new RuntimeException("Must have at least 1 worker thread.");
          }
          ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat("dr-el-executor-thread-%d").build();
          _threadPoolExecutor = new PriorityBasedThreadPoolExecutor(_executorNum, _executorNum, 0L,
              TimeUnit.MILLISECONDS, factory);

          boolean firstRun = true;
          while (_running.get() && !Thread.currentThread().isInterrupted()) {
            _analyticJobGenerator.updateResourceManagerAddresses();
            lastRun = System.currentTimeMillis();

            logger.info("Fetching analytic job list...");

            try {
              _hadoopSecurity.checkLogin();
            } catch (IOException e) {
              logger.info("Error with hadoop kerberos login", e);
              //Wait for a while before retry
              waitInterval(_retryInterval);
              continue;
            }

            List<AnalyticJob> todos;
            try {
              todos = _analyticJobGenerator.fetchAnalyticJobs();
            } catch (Exception e) {
              logger.error("Error fetching job list. Try again later...", e);
              //Wait for a while before retry
              waitInterval(_retryInterval);
              continue;
            }

            // Submit jobs fetched from RM and get the lowest finish time amongst the apps fetched from RM on first run.
            long lowestFinishTime = Long.MAX_VALUE;
            String appId = "";
            for (AnalyticJob analyticJob : todos) {
              if (firstRun && analyticJob.getFinishTime() < lowestFinishTime) {
                lowestFinishTime = analyticJob.getFinishTime();
                appId = analyticJob.getAppId();
              }
              synchronized (_appsLock) {
                Future<?> future = _threadPoolExecutor.submit(RunnableWithPriority.get(new ExecutorJob(analyticJob)));
                analyticJob.setJobFuture(future);
                addToFinishTimesMap(analyticJob.getAppType().getName(), analyticJob.getFinishTime());
                _appToAnalyticJobMap.put(analyticJob.getAppId(), analyticJob);
              }
            }

            // On first run, if backfill is enabled, launch a backfill thread to get and submit a list of apps to be
            // backfilled.
            if (firstRun && _backfillEnabled) {
              logger.info("Lowest finish time retrieved from RM is " +
                  DATE_FORMAT_GMT.format(new Date(lowestFinishTime)) + " for " + appId);
              List<BackfillData> backfillDataList = populateAndGetBackfillInfoData(lowestFinishTime);
              if (backfillDataList != null && !backfillDataList.isEmpty()) {
                _backFillThread = new Thread(new BackfillThread(backfillDataList, lowestFinishTime), "Backfill Thread");
                _backFillThread.start();
              } else {
                logger.info("Backfill is not required.");
              }
            }
            firstRun = false;
            int queueSize = _threadPoolExecutor.getQueue().size();
            MetricsController.setQueueSize(queueSize);
            logger.info("Job queue size is " + queueSize);

            //Wait for a while before next fetch
            waitInterval(_fetchInterval);
          }
          logger.info("Main thread is terminated.");
          return null;
        }
The class that runs the Dr. Elephant daemon
app/com/linkedin/drelephant/ElephantRunner.java