AgentConsumer extends Stoppable { public static void cleanupPreviouslyRunningJobs( JobDao dao, Map<Long, CallableJob> jobRuns) { for (Entry<Long, CallableJob> job : jobRuns.entrySet()) { CallableJob value = job.getValue(); value.getFinish().set(System.currentTimeMillis()); dao.updateJobRun(value); } } AgentConsumer(JobDao dao, Reporting reporter,
      String hostname, MailInfo mailInfo, Session session,
      List<SupportedDriver> drivers, int numOfConcurrentJobs,
      int numOfConcurrentReruns, int maxReruns, long waitBeforeRetrySeconds,
      int minAttemptsForNotification); void writeReportToLocal(String reportRootPath); String getReportRootPath(); void init(); static void cleanupPreviouslyRunningJobs(
      JobDao dao, Map<Long, CallableJob> jobRuns); void doRun(); CallableJob assembleCallableJob(PlannedJob plannedJob,
      int attemptNumber); void submitJob(final CallableJob cj); @Override void run(); @Override void close(); static void setShouldSendErrorReports(boolean shouldSend); @CoverageIgnore static void sendErrorReport(JobSpec jobSpec, String query,
                                     Exception ex, Long myId, String hostname,
                                     MailInfo mailInfo, Session session,
                                     int attemptNumber); synchronized Map<Long, CallableJob> getFinishedJobs(int limit); static boolean isJobFailed(CallableJob cj); synchronized Map<Long, CallableJob> getFailedQueries(int limit); synchronized Map<Long, CallableJob> getSuccesfulQueries(int limit); synchronized CallableJob getLatestMatching(PlannedJob pj, int limit); static Logger LOG; public int SLEEP_FOR; final ThreadPoolExecutor executor; static int maxReruns; final static int LIMIT_JOB_RUNS; final static int START_ATTEMPT_NUM; }
@Test public void testCleanupPreviouslyRunningJobs() { JobSpec aJob = TestAgent.getTestJob("Mary Wollstonecraft", dao); long id = dao.createJob(aJob); PlannedJob pj = new PlannedJob(dao.getJob(id), Utils.getCurrentTime()); SleepyCallableQuery cj = new SleepyCallableQuery( pj, dao, reporting, "example.com", mailInfo, null, null, drivers.get(0), 1, 10000); consumer.submitJob(cj); doSleep(); assertEquals(0, consumer.getFailedQueries(limit).size()); assertEquals(1, dao.getRunningJobs().size()); AgentConsumer.cleanupPreviouslyRunningJobs(dao, dao.getRunningJobs()); assertEquals(0, dao.getRunningJobs().size()); assertEquals(1, consumer.getFailedQueries(limit).size()); }