AuroraCronJob implements Job { @VisibleForTesting void doExecute(final IJobKey key) throws JobExecutionException { final String path = JobKeys.canonicalString(key); final Optional<DeferredLaunch> deferredLaunch = storage.write( (MutateWork.Quiet<Optional<DeferredLaunch>>) storeProvider -> { Optional<IJobConfiguration> config = storeProvider.getCronJobStore().fetchJob(key); if (!config.isPresent()) { LOG.warn( "Cron was triggered for {} but no job with that key was found in storage.", path); CRON_JOB_MISFIRES.incrementAndGet(); return Optional.absent(); } SanitizedCronJob cronJob; try { cronJob = SanitizedCronJob.fromUnsanitized(configurationManager, config.get()); } catch (ConfigurationManager.TaskDescriptionException | CronException e) { LOG.warn( "Invalid cron job for {} in storage - failed to parse with {}", key, e); CRON_JOB_PARSE_FAILURES.incrementAndGet(); return Optional.absent(); } CronCollisionPolicy collisionPolicy = cronJob.getCronCollisionPolicy(); LOG.info( "Cron triggered for {} at {} with policy {}", path, new Date(), collisionPolicy); CRON_JOB_TRIGGERS.incrementAndGet(); final Query.Builder activeQuery = Query.jobScoped(key).active(); Set<String> activeTasks = Tasks.ids(storeProvider.getTaskStore().fetchTasks(activeQuery)); ITaskConfig task = cronJob.getSanitizedConfig().getJobConfig().getTaskConfig(); Set<Integer> instanceIds = cronJob.getSanitizedConfig().getInstanceIds(); if (activeTasks.isEmpty()) { stateManager.insertPendingTasks(storeProvider, task, instanceIds); return Optional.absent(); } CRON_JOB_COLLISIONS.incrementAndGet(); switch (collisionPolicy) { case KILL_EXISTING: return Optional.of(new DeferredLaunch(task, instanceIds, activeTasks)); case RUN_OVERLAP: LOG.error("Ignoring trigger for job {} with deprecated collision" + "policy RUN_OVERLAP due to unterminated active tasks.", path); return Optional.absent(); case CANCEL_NEW: return Optional.absent(); default: LOG.error("Unrecognized cron collision policy: " + collisionPolicy); return Optional.absent(); } } ); if (!deferredLaunch.isPresent()) { return; } storage.write((NoResult.Quiet) storeProvider -> { for (String taskId : deferredLaunch.get().activeTaskIds) { stateManager.changeState( storeProvider, taskId, Optional.absent(), KILLING, KILL_AUDIT_MESSAGE); } }); LOG.info("Waiting for job to terminate before launching cron job {}.", path); final Query.Builder query = Query.taskScoped(deferredLaunch.get().activeTaskIds).active(); try { delayedStartBackoff.doUntilSuccess(() -> { if (Iterables.isEmpty(Storage.Util.fetchTasks(storage, query))) { LOG.info("Initiating delayed launch of cron " + path); storage.write((NoResult.Quiet) storeProvider -> stateManager.insertPendingTasks( storeProvider, deferredLaunch.get().task, deferredLaunch.get().instanceIds)); return true; } else { LOG.info("Not yet safe to run cron " + path); return false; } }); } catch (InterruptedException e) { LOG.warn("Interrupted while trying to launch cron " + path, e); Thread.currentThread().interrupt(); throw new JobExecutionException(e); } } @Inject  AuroraCronJob(
      ConfigurationManager configurationManager,
      Config config,
      Storage storage,
      StateManager stateManager); @Override void execute(JobExecutionContext context);  }
@Test public void testExecuteNonexistentIsNoop() throws JobExecutionException { control.replay(); auroraCronJob.doExecute(QuartzTestUtil.AURORA_JOB_KEY); }
@Test public void testEmptyStorage() throws JobExecutionException { stateManager.insertPendingTasks( EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject()); expectLastCall().times(3); control.replay(); populateStorage(CronCollisionPolicy.CANCEL_NEW); auroraCronJob.doExecute(QuartzTestUtil.AURORA_JOB_KEY); storage = DbUtil.createStorage(); populateStorage(CronCollisionPolicy.KILL_EXISTING); auroraCronJob.doExecute(QuartzTestUtil.AURORA_JOB_KEY); storage = DbUtil.createStorage(); populateStorage(CronCollisionPolicy.RUN_OVERLAP); auroraCronJob.doExecute(QuartzTestUtil.AURORA_JOB_KEY); }
@Test public void testCancelNew() throws JobExecutionException { control.replay(); populateTaskStore(); populateStorage(CronCollisionPolicy.CANCEL_NEW); auroraCronJob.doExecute(QuartzTestUtil.AURORA_JOB_KEY); }
@Test public void testKillExisting() throws Exception { Capture<ExceptionalSupplier<Boolean, RuntimeException>> capture = createCapture(); expect(stateManager.changeState( EasyMock.anyObject(), eq(TASK_ID), eq(Optional.absent()), eq(ScheduleStatus.KILLING), eq(AuroraCronJob.KILL_AUDIT_MESSAGE))) .andReturn(StateChangeResult.SUCCESS); backoffHelper.doUntilSuccess(EasyMock.capture(capture)); stateManager.insertPendingTasks( EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject()); control.replay(); populateStorage(CronCollisionPolicy.KILL_EXISTING); populateTaskStore(); auroraCronJob.doExecute(QuartzTestUtil.AURORA_JOB_KEY); assertFalse(capture.getValue().get()); storage.write( (NoResult.Quiet) storeProvider -> storeProvider.getUnsafeTaskStore().deleteAllTasks()); assertTrue(capture.getValue().get()); }