public void run() {
        populateTestData();
        long backfillTs = 1526400000000L;
        List<AppResult> results = AppResult.find.findList();
        int initialResultCount = results.size();
        BackfillInfo mapReduceBackfillInfo = BackfillInfo.find.byId("MAPREDUCE");
        assertNotNull(mapReduceBackfillInfo);
        mapReduceBackfillInfo.backfillTs = backfillTs;
        mapReduceBackfillInfo.save();
        BackfillInfo sparkBackfillInfo = BackfillInfo.find.byId("SPARK");
        assertNotNull(sparkBackfillInfo);
        sparkBackfillInfo.backfillTs = backfillTs;
        sparkBackfillInfo.save();
        BackfillInfo tezBackfillInfo = new BackfillInfo();
        tezBackfillInfo.appType = "TEZ";
        tezBackfillInfo.backfillTs = backfillTs;
        tezBackfillInfo.save();
        List<BackfillInfo> infos = BackfillInfo.find.findList();
        assertEquals(3, infos.size());

        // Enable backfill and start runner with 1 thread.
        ElephantContext.instance().getGeneralConf().setInt("drelephant.analysis.thread.count", 1);
        ElephantContext.instance().getGeneralConf().setBoolean("drelephant.analysis.backfill.enabled", true);
        final ElephantRunner runner = spy(new ElephantRunner());
        Map<String, Priority> appsToPriorityMap = Collections.synchronizedMap(new LinkedHashMap<String, Priority>());
        // 2 jobs to be returned by job generator.
        // For each job, set job execution priority as MIN_PRIORITY. This acts as a sentinel value and we will wait
        // during analysis till the value changes to avoid race while adding to appsToPriorityMap.
        List<AnalyticJob> jobs = new ArrayList<AnalyticJob>();
        jobs.add(new DummyAnalyticJobGenerator.DummyAnalyticJob().setAppId("application_1526555215992_0007").
            setAppType(new ApplicationType("MAPREDUCE")).setName("mapreduce_job_3").setQueueName("default").
            setStartTime(1526800000000L).setFinishTime(1526800000050L).setUser("user"));
        jobs.add(new DummyAnalyticJobGenerator.DummyAnalyticJob().setAppId("application_1526555215992_0008").
            setAppType(new ApplicationType("SPARK")).setName("spark_job_3").setQueueName("default").
            setStartTime(1526900000000L).setFinishTime(1526900000050L).setUser("user"));
        long lowestTsFromRM = 1526800000050L;
        AnalyticJobGenerator jobGenerator = new DummyAnalyticJobGenerator(appsToPriorityMap, jobs);
        doReturn(jobGenerator).when(runner).getAnalyticJobGenerator();

        // 6 jobs to be backfilled.
        List<AnalyticJob> mapReduceBackfillJobs = new ArrayList<AnalyticJob>();
        mapReduceBackfillJobs.add(new DummyAnalyticJobGenerator.DummyAnalyticJob().
            setAppId("application_1526555215992_0001").setAppType(new ApplicationType("MAPREDUCE")).
            setName("mapreduce_job_1").setQueueName("default").setStartTime(1526510000000L).
            setFinishTime(1526520000150L).setUser("user").setJobExecutionPriority(Priority.MIN_PRIORITY));
        mapReduceBackfillJobs.add(new DummyAnalyticJobGenerator.DummyAnalyticJob().
            setAppId("application_1526555215992_0003").setAppType(new ApplicationType("MAPREDUCE")).
            setName("mapreduce_job_2").setQueueName("default").setStartTime(1526520000000L).
            setFinishTime(1526520000050L).setUser("user").setJobExecutionPriority(Priority.MIN_PRIORITY));
        DummyElephantBackfillFetcher mapreduceBackfillFetcher =
            new DummyElephantBackfillFetcher(mapReduceBackfillJobs, "MAPREDUCE", backfillTs, lowestTsFromRM);
        when(runner.getBackfillFetcher(eq(new ApplicationType("MAPREDUCE")))).thenReturn(mapreduceBackfillFetcher);

        List<AnalyticJob> sparkBackfillJobs = new ArrayList<AnalyticJob>();
        sparkBackfillJobs.add(new DummyAnalyticJobGenerator.DummyAnalyticJob().setAppId("application_1526555215992_0002").
            setAppType(new ApplicationType("SPARK")).setName("spark_job_1").setQueueName("default").
            setStartTime(1526510050000L).setFinishTime(1526700000060L).setUser("user").
            setJobExecutionPriority(Priority.MIN_PRIORITY));
        sparkBackfillJobs.add(new DummyAnalyticJobGenerator.DummyAnalyticJob().setAppId("application_1526555215992_0006").
            setAppType(new ApplicationType("SPARK")).setName("spark_job_2").setQueueName("default").
            setStartTime(1526700000000L).setFinishTime(1526700000050L).setUser("user").
            setJobExecutionPriority(Priority.MIN_PRIORITY));
        DummyElephantBackfillFetcher sparkBackfillFetcher =
            new DummyElephantBackfillFetcher(sparkBackfillJobs, "SPARK", backfillTs, lowestTsFromRM);
        when(runner.getBackfillFetcher(eq(new ApplicationType("SPARK")))).thenReturn(sparkBackfillFetcher);

        // 2 jobs below will not have finish time at the time of backfill.
        List<AnalyticJob> tezBackfillJobs = new ArrayList<AnalyticJob>();
        tezBackfillJobs.add(
            ((DummyAnalyticJobGenerator.DummyAnalyticJob) new DummyAnalyticJobGenerator.DummyAnalyticJob().
            setAppId("application_1526555215992_0004").setAppType(new ApplicationType("TEZ")).
            setName("tez_job_1").setQueueName("default").setUser("user").setStartTime(1526600000000L)).
            setFinishTimeAfterAnalysis(1526600000050L).setJobExecutionPriority(Priority.MIN_PRIORITY));
        tezBackfillJobs.add(
            ((DummyAnalyticJobGenerator.DummyAnalyticJob) new DummyAnalyticJobGenerator.DummyAnalyticJob().
            setAppId("application_1526555215992_0005").setAppType(new ApplicationType("TEZ")).
            setName("tez_job_2").setQueueName("default").setUser("user").setStartTime(1526600000020L)).
            setFinishTimeAfterAnalysis(1526600000040L).setJobExecutionPriority(Priority.MIN_PRIORITY));
        ApplicationType tezAppType = new ApplicationType("TEZ");
        doReturn(tezAppType).when(runner).getAppTypeForName("TEZ");
        DummyElephantBackfillFetcher tezBackfillFetcher = new DummyElephantBackfillFetcher(
            tezBackfillJobs, "TEZ", backfillTs, lowestTsFromRM);
        when(runner.getBackfillFetcher(eq(tezAppType))).thenReturn(tezBackfillFetcher);

        Thread runnerThread = new Thread(new Runnable() {
          @Override
          public void run() {
            runner.run();
          }
        });
        runnerThread.start();

        // Wait till results have been updated in DB.
        results = AppResult.find.findList();
        int resultCount = results.size();
        int loopCount = 400;
        while (resultCount < 8 + initialResultCount && loopCount > 0) {
          try {
            Thread.sleep(50);
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
          results = AppResult.find.findList();
          resultCount = results.size();
          loopCount--;
        }

        // Check yarn_app_result.
        assertEquals(8 + initialResultCount, resultCount);
        assertAppResultEntry("application_1526555215992_0001", 1526510000000L, 1526520000150L,
            "mapreduce_job_1");
        assertAppResultEntry("application_1526555215992_0002", 1526510050000L, 1526700000060L,
            "spark_job_1");
        assertAppResultEntry("application_1526555215992_0003", 1526520000000L, 1526520000050L,
            "mapreduce_job_2");
        assertAppResultEntry("application_1526555215992_0004", 1526600000000L, 1526600000050L,
            "tez_job_1");
        assertAppResultEntry("application_1526555215992_0005", 1526600000020L, 1526600000040L,
            "tez_job_2");
        assertAppResultEntry("application_1526555215992_0006", 1526700000000L, 1526700000050L,
            "spark_job_2");
        assertAppResultEntry("application_1526555215992_0007", 1526800000000L, 1526800000050L,
            "mapreduce_job_3");
        assertAppResultEntry("application_1526555215992_0008", 1526900000000L, 1526900000050L,
            "spark_job_3");
        DummyElephantBackfillFetcher[] backfillFetchers =
            new DummyElephantBackfillFetcher[] {mapreduceBackfillFetcher, sparkBackfillFetcher, tezBackfillFetcher};
        for (DummyElephantBackfillFetcher backfillFetcher : backfillFetchers) {
          try {
            assertTrue("Jobs for backfill (of type " + backfillFetcher.getFetcherName() + ") should have been "
                + "fetched based on lowest ts of apps retrieved from RM and backfill ts in DB",
                backfillFetcher.getJobsFetchedFlag());
          } catch (Exception e) {
            fail("Unexpected exception.");
          }
        }

        // Verify apps submitted.
        assertEquals(8, appsToPriorityMap.size());
        assertEquals(Lists.newArrayList(Priority.NORMAL, Priority.NORMAL, Priority.LOW, Priority.LOW, Priority.LOW,
            Priority.LOW, Priority.LOW, Priority.LOW), new ArrayList<Priority>(appsToPriorityMap.values()));

        String[] appTypes = new String[] {"MAPREDUCE", "SPARK", "TEZ"};
        for (String appType : appTypes) {
          ElephantRunner.FinishTimeInfo finishTimeInfo = runner.getFinishTimeInfo(appType);
          loopCount = 400;
          while ((finishTimeInfo == null || !finishTimeInfo._appFinishTimesMap.isEmpty()) && loopCount > 0) {
            try {
              Thread.sleep(50);
            } catch (InterruptedException e) {
              e.printStackTrace();
            }
            loopCount--;
          }
        }

        // Ensure cleanup is performed.
        assertTrue("App to analytic map should have been empty", runner.getAppToAnalyticJobMap().isEmpty());
        assertNotNull("MAPREDUCE app type should have final info", runner.getFinishTimeInfo("MAPREDUCE"));
        assertTrue("Finish times map should be cleaned up for MAPREDUCE app type",
            runner.getFinishTimeInfo("MAPREDUCE")._appFinishTimesMap.isEmpty());
        assertNotNull("SPARK app type should have final info", runner.getFinishTimeInfo("SPARK"));
        assertTrue("Finish times map should be cleaned up for SPARK app type",
            runner.getFinishTimeInfo("SPARK")._appFinishTimesMap.isEmpty());
        assertNotNull("TEZ app type should have final info", runner.getFinishTimeInfo("TEZ"));
        assertTrue("Finish times map should be cleaned up for TEZ app type",
            runner.getFinishTimeInfo("TEZ")._appFinishTimesMap.isEmpty());

        // Check backfill_info.
        assertEquals(3, BackfillInfo.find.findList().size());
        assertBackfillInfo("MAPREDUCE", 1526800000050L);
        assertBackfillInfo("SPARK", 1526900000050L);

        // Stop the runner and do cleanup before exiting.
        runner.kill();
        runnerThread.interrupt();
        try {
          runnerThread.join();
        } catch (InterruptedException e) {
          // Ignore the exception.
        }
      }
Tests backfilling of jobs when Dr.Elephant comes up.
test/com/linkedin/drelephant/BackfillTest.java