public void run() {
        List<BackfillInfo> infos = BackfillInfo.find.findList();
        assertEquals(0, infos.size());

        // Start runner with 2 threads.
        ElephantContext.instance().getGeneralConf().setInt("drelephant.analysis.thread.count", 2);
        final ElephantRunner runner = spy(new ElephantRunner());
        Map<String, Priority> appsToPriorityMap = Collections.synchronizedMap(new LinkedHashMap<String, Priority>());
        // 4 jobs to be returned.
        List<AnalyticJob> jobs = new ArrayList<AnalyticJob>();
        jobs.add(new DummyAnalyticJobGenerator.DummyAnalyticJob().setAppId("application_1526555215992_0001").
            setAppType(new ApplicationType("MAPREDUCE")).setName("mapreduce_job").setQueueName("default").
            setStartTime(1526520000000L).setFinishTime(1526520000050L).setUser("user"));
        jobs.add(new DummyAnalyticJobGenerator.DummyAnalyticJob().setAppId("application_1526555215992_0002").
            setAppType(new ApplicationType("SPARK")).setName("spark_job").setQueueName("default").
            setStartTime(1526600000000L).setFinishTime(1526600000050L).setUser("user"));
        jobs.add(new DummyAnalyticJobGenerator.DummyAnalyticJob().setAppId("application_1526555215992_0003").
            setAppType(new ApplicationType("SPARK")).setName("spark_job").setQueueName("default").
            setStartTime(1526700000000L).setFinishTime(1526700000050L).setUser("user"));
        jobs.add(new DummyAnalyticJobGenerator.DummyAnalyticJob().setAppId("application_1526555215992_0004").
            setAppType(new ApplicationType("MAPREDUCE")).setName("mapreduce_job").setQueueName("default").
            setStartTime(1526800000000L).setFinishTime(1526800000050L).setUser("user"));
        doReturn(new DummyAnalyticJobGenerator(appsToPriorityMap, jobs)).when(runner).getAnalyticJobGenerator();
        Thread runnerThread = new Thread(new Runnable() {
          @Override
          public void run() {
            runner.run();
          }
        });
        runnerThread.start();

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

        // Check yarn_app_result.
        assertEquals(4, resultCount);
        assertAppResultEntry("application_1526555215992_0001", 1526520000000L, 1526520000050L,
            "mapreduce_job");
        assertAppResultEntry("application_1526555215992_0002", 1526600000000L, 1526600000050L,
            "spark_job");
        assertAppResultEntry("application_1526555215992_0003", 1526700000000L, 1526700000050L,
            "spark_job");
        assertAppResultEntry("application_1526555215992_0004", 1526800000000L, 1526800000050L,
            "mapreduce_job");

        // Check backfill_info.
        assertEquals(2, BackfillInfo.find.findList().size());
        BackfillInfo mapReduceInfo = BackfillInfo.find.byId("MAPREDUCE");
        assertEquals(1526800000050L, mapReduceInfo.backfillTs);
        BackfillInfo sparkInfo = BackfillInfo.find.byId("SPARK");
        assertEquals(1526700000050L, sparkInfo.backfillTs);

        // Verify if apps were submitted with NORMAL priority.
        assertEquals(4, appsToPriorityMap.size());
        for (Priority priority : appsToPriorityMap.values()) {
          assertEquals(Priority.NORMAL, priority);
        }

        // 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());

        // 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