@Test public void testExceptionReporting() throws InterruptedException { final AtomicBoolean exceptionWasThrown = new AtomicBoolean(false); final OneShotLatch latch = new OneShotLatch(); final Object lock = new Object(); ProcessingTimeService timeServiceProvider = new SystemProcessingTimeService( new AsyncExceptionHandler() { @Override public void handleAsyncException(String message, Throwable exception) { exceptionWasThrown.set(true); latch.trigger(); } }, lock); timeServiceProvider.registerTimer(System.currentTimeMillis(), new ProcessingTimeCallback() { @Override public void onProcessingTime(long timestamp) throws Exception { throw new Exception("Exception in Timer"); } }); latch.await(); assertTrue(exceptionWasThrown.get()); }
@Override public ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback target) { long delay = Math.max(timestamp - getCurrentProcessingTime(), 0) + 1; try { return timerService.schedule( new TriggerTask(status, task, checkpointLock, target, timestamp), delay, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException e) { final int status = this.status.get(); if (status == STATUS_QUIESCED) { return new NeverCompleteFuture(delay); } else if (status == STATUS_SHUTDOWN) { throw new IllegalStateException("Timer service is shut down"); } else { throw e; } } }
SystemProcessingTimeService extends ProcessingTimeService { @Override public ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback target) { long delay = Math.max(timestamp - getCurrentProcessingTime(), 0) + 1; try { return timerService.schedule( new TriggerTask(status, task, checkpointLock, target, timestamp), delay, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException e) { final int status = this.status.get(); if (status == STATUS_QUIESCED) { return new NeverCompleteFuture(delay); } else if (status == STATUS_SHUTDOWN) { throw new IllegalStateException("Timer service is shut down"); } else { throw e; } } } }
SystemProcessingTimeService extends ProcessingTimeService { @Override public ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback target) { long delay = Math.max(timestamp - getCurrentProcessingTime(), 0) + 1; try { return timerService.schedule( new TriggerTask(status, task, checkpointLock, target, timestamp), delay, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException e) { final int status = this.status.get(); if (status == STATUS_QUIESCED) { return new NeverCompleteFuture(delay); } else if (status == STATUS_SHUTDOWN) { throw new IllegalStateException("Timer service is shut down"); } else { throw e; } } } SystemProcessingTimeService(AsyncExceptionHandler failureHandler, Object checkpointLock);  SystemProcessingTimeService(
			AsyncExceptionHandler task,
			Object checkpointLock,
			ThreadFactory threadFactory); }
SystemProcessingTimeService extends ProcessingTimeService { @Override public ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback target) { long delay = Math.max(timestamp - getCurrentProcessingTime(), 0) + 1; try { return timerService.schedule( new TriggerTask(status, task, checkpointLock, target, timestamp), delay, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException e) { final int status = this.status.get(); if (status == STATUS_QUIESCED) { return new NeverCompleteFuture(delay); } else if (status == STATUS_SHUTDOWN) { throw new IllegalStateException("Timer service is shut down"); } else { throw e; } } } SystemProcessingTimeService(AsyncExceptionHandler failureHandler, Object checkpointLock);  SystemProcessingTimeService(
			AsyncExceptionHandler task,
			Object checkpointLock,
			ThreadFactory threadFactory); @Override long getCurrentProcessingTime(); @Override ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback target); @Override ScheduledFuture<?> scheduleAtFixedRate(ProcessingTimeCallback callback, long initialDelay, long period); @Override boolean isTerminated(); @Override void quiesce(); @Override void awaitPendingAfterQuiesce(); @Override void shutdownService(); @Override boolean shutdownAndAwaitPending(long time, TimeUnit timeUnit); @Override boolean shutdownServiceUninterruptible(long timeoutMs); }
SystemProcessingTimeService extends ProcessingTimeService { @Override public ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback target) { long delay = Math.max(timestamp - getCurrentProcessingTime(), 0) + 1; try { return timerService.schedule( new TriggerTask(status, task, checkpointLock, target, timestamp), delay, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException e) { final int status = this.status.get(); if (status == STATUS_QUIESCED) { return new NeverCompleteFuture(delay); } else if (status == STATUS_SHUTDOWN) { throw new IllegalStateException("Timer service is shut down"); } else { throw e; } } } SystemProcessingTimeService(AsyncExceptionHandler failureHandler, Object checkpointLock);  SystemProcessingTimeService(
			AsyncExceptionHandler task,
			Object checkpointLock,
			ThreadFactory threadFactory); @Override long getCurrentProcessingTime(); @Override ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback target); @Override ScheduledFuture<?> scheduleAtFixedRate(ProcessingTimeCallback callback, long initialDelay, long period); @Override boolean isTerminated(); @Override void quiesce(); @Override void awaitPendingAfterQuiesce(); @Override void shutdownService(); @Override boolean shutdownAndAwaitPending(long time, TimeUnit timeUnit); @Override boolean shutdownServiceUninterruptible(long timeoutMs);  }