@Test
  public void testPrioritizeExecution() {
    running(testServer(TEST_SERVER_PORT, fakeApp), new Runnable() {
      public void run() {
        List<BackfillInfo> infos = BackfillInfo.find.findList();
        assertEquals(0, infos.size());

        // Start runner with 1 thread.
        ElephantContext.instance().getGeneralConf().setInt("drelephant.analysis.thread.count", 1);
        final ElephantRunner runner = spy(new ElephantRunner());
        Map<String, Priority> appsToPriorityMap = Collections.synchronizedMap(new LinkedHashMap<String, Priority>());
        // 3 jobs to be returned.
        List<AnalyticJob> jobs = new ArrayList<AnalyticJob>();
        // First job will sleep for 5 seconds before analysis, if not interrupted.
        jobs.add(((DummyAnalyticJobGenerator.DummyAnalyticJob) new DummyAnalyticJobGenerator.DummyAnalyticJob().
            setAppId("application_1526555215992_0001").setAppType(new ApplicationType("MAPREDUCE")).
            setName("mapreduce_job").setQueueName("default").setStartTime(1526520000000L).setFinishTime(1526520000050L).
            setUser("user")).setSleepBeforeAnalysis());
        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"));
        doReturn(new DummyAnalyticJobGenerator(appsToPriorityMap, jobs)).when(runner).getAnalyticJobGenerator();
        Thread runnerThread = new Thread(new Runnable() {
          @Override
          public void run() {
            runner.run();
          }
        });
        runnerThread.start();

        // Wait till first app has been picked up by executor.
        int loopCount = 400;
        while (!runner.getAppToAnalyticJobMap().containsKey("application_1526555215992_0001") && loopCount > 0) {
          try {
            Thread.sleep(50);
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
          loopCount--;
        }
        assertTrue("Executor should have picked up application_1526555215992_0001 for execution",
            runner.getAppToAnalyticJobMap().containsKey("application_1526555215992_0001"));

        // Prioritize application_1526555215992_0003.
        Future<?> future = runner.prioritizeExecutionAndReturnFuture("application_1526555215992_0003");
        assertNotNull("Priortization was not successful", future);
        // Ensure first job proceeds ahead from sleep by interrupting it.
        runner.getAppToAnalyticJobMap().get("application_1526555215992_0001").getJobFuture().cancel(true);
        runner.waitTillPrioritizedJobAnalysisFinishes(future);
        assertTrue("Prioritized job should have finished.", future.isDone());

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

        // Check yarn_app_result.
        assertEquals("yarn_app_result table should have 3 rows", 3, 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");

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

        // Verify if apps were submitted in correct order based on prioritization.
        assertEquals(3, appsToPriorityMap.size());
        assertEquals(Lists.newArrayList("application_1526555215992_0001", "application_1526555215992_0003",
            "application_1526555215992_0002"), new ArrayList<String>(appsToPriorityMap.keySet()));
        assertEquals(Lists.newArrayList(Priority.NORMAL, Priority.HIGH, Priority.NORMAL),
            new ArrayList<Priority>(appsToPriorityMap.values()));

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