@Test public void testTaskAndJobPredicate() { SettableTaskProgress t1 = new SettableTaskProgress("j1", "t1", WorkUnitState.WorkingState.RUNNING, false); SettableTaskProgress t2 = new SettableTaskProgress("j1", "t1", WorkUnitState.WorkingState.RUNNING, false); SettableJobProgress jobProgress = new SettableJobProgress("job123", 10, 0, 0, JobState.RunningState.RUNNING, Lists.newArrayList(t1, t2)); AtomicBoolean atomicBoolean = new AtomicBoolean(false); JobInterruptionPredicate predicate = new JobInterruptionPredicate(jobProgress, "SELECT EXISTS(SELECT * FROM (SELECT completedTasks > 5 AS pred FROM jobProgress UNION SELECT count(*) > 0 AS pred FROM taskProgress WHERE workingState = 'FAILED') WHERE pred)", () -> atomicBoolean.set(true), false); predicate.runOneIteration(); Assert.assertFalse(atomicBoolean.get()); t2.workingState = WorkUnitState.WorkingState.FAILED; predicate.runOneIteration(); Assert.assertTrue(atomicBoolean.get()); atomicBoolean.set(false); t2.workingState = WorkUnitState.WorkingState.RUNNING; predicate.runOneIteration(); Assert.assertFalse(atomicBoolean.get()); jobProgress.completedTasks = 6; predicate.runOneIteration(); Assert.assertTrue(atomicBoolean.get()); }
@Override protected void runOneIteration() { if (this.evaluator == null) { stopAsync(); return; } switch (this.jobProgress.getState()) { case PENDING: return; case RUNNING: try { List<Object> objects = Stream.concat(Stream.<Object>of(this.jobProgress), this.jobProgress.getTaskProgress().stream()).collect( Collectors.toList()); if (this.evaluator.evaluate(objects)) { log.info("Interrupting job due to satisfied job interruption predicate. Predicate: " + this.sql); this.jobInterruptionHook.run(); stopAsync(); } } catch (Throwable exc) { log.warn("Failed to evaluate job interruption predicate. Will not preemptively interrupt job.", exc); throw Throwables.propagate(exc); } break; default: log.info(String.format("Detected job finished with state %s. Stopping job interruption predicate.", this.jobProgress.getState())); stopAsync(); } }
JobInterruptionPredicate extends AbstractScheduledService { @Override protected void runOneIteration() { if (this.evaluator == null) { stopAsync(); return; } switch (this.jobProgress.getState()) { case PENDING: return; case RUNNING: try { List<Object> objects = Stream.concat(Stream.<Object>of(this.jobProgress), this.jobProgress.getTaskProgress().stream()).collect( Collectors.toList()); if (this.evaluator.evaluate(objects)) { log.info("Interrupting job due to satisfied job interruption predicate. Predicate: " + this.sql); this.jobInterruptionHook.run(); stopAsync(); } } catch (Throwable exc) { log.warn("Failed to evaluate job interruption predicate. Will not preemptively interrupt job.", exc); throw Throwables.propagate(exc); } break; default: log.info(String.format("Detected job finished with state %s. Stopping job interruption predicate.", this.jobProgress.getState())); stopAsync(); } } }
JobInterruptionPredicate extends AbstractScheduledService { @Override protected void runOneIteration() { if (this.evaluator == null) { stopAsync(); return; } switch (this.jobProgress.getState()) { case PENDING: return; case RUNNING: try { List<Object> objects = Stream.concat(Stream.<Object>of(this.jobProgress), this.jobProgress.getTaskProgress().stream()).collect( Collectors.toList()); if (this.evaluator.evaluate(objects)) { log.info("Interrupting job due to satisfied job interruption predicate. Predicate: " + this.sql); this.jobInterruptionHook.run(); stopAsync(); } } catch (Throwable exc) { log.warn("Failed to evaluate job interruption predicate. Will not preemptively interrupt job.", exc); throw Throwables.propagate(exc); } break; default: log.info(String.format("Detected job finished with state %s. Stopping job interruption predicate.", this.jobProgress.getState())); stopAsync(); } } JobInterruptionPredicate(JobState jobState, Runnable jobInterruptionHook, boolean autoStart); protected  JobInterruptionPredicate(JobProgress jobProgress, String predicate,
			Runnable jobInterruptionHook, boolean autoStart); }
JobInterruptionPredicate extends AbstractScheduledService { @Override protected void runOneIteration() { if (this.evaluator == null) { stopAsync(); return; } switch (this.jobProgress.getState()) { case PENDING: return; case RUNNING: try { List<Object> objects = Stream.concat(Stream.<Object>of(this.jobProgress), this.jobProgress.getTaskProgress().stream()).collect( Collectors.toList()); if (this.evaluator.evaluate(objects)) { log.info("Interrupting job due to satisfied job interruption predicate. Predicate: " + this.sql); this.jobInterruptionHook.run(); stopAsync(); } } catch (Throwable exc) { log.warn("Failed to evaluate job interruption predicate. Will not preemptively interrupt job.", exc); throw Throwables.propagate(exc); } break; default: log.info(String.format("Detected job finished with state %s. Stopping job interruption predicate.", this.jobProgress.getState())); stopAsync(); } } JobInterruptionPredicate(JobState jobState, Runnable jobInterruptionHook, boolean autoStart); protected  JobInterruptionPredicate(JobProgress jobProgress, String predicate,
			Runnable jobInterruptionHook, boolean autoStart);  }
JobInterruptionPredicate extends AbstractScheduledService { @Override protected void runOneIteration() { if (this.evaluator == null) { stopAsync(); return; } switch (this.jobProgress.getState()) { case PENDING: return; case RUNNING: try { List<Object> objects = Stream.concat(Stream.<Object>of(this.jobProgress), this.jobProgress.getTaskProgress().stream()).collect( Collectors.toList()); if (this.evaluator.evaluate(objects)) { log.info("Interrupting job due to satisfied job interruption predicate. Predicate: " + this.sql); this.jobInterruptionHook.run(); stopAsync(); } } catch (Throwable exc) { log.warn("Failed to evaluate job interruption predicate. Will not preemptively interrupt job.", exc); throw Throwables.propagate(exc); } break; default: log.info(String.format("Detected job finished with state %s. Stopping job interruption predicate.", this.jobProgress.getState())); stopAsync(); } } JobInterruptionPredicate(JobState jobState, Runnable jobInterruptionHook, boolean autoStart); protected  JobInterruptionPredicate(JobProgress jobProgress, String predicate,
			Runnable jobInterruptionHook, boolean autoStart);  static final String INTERRUPTION_SQL; }