InsertWatermarksP extends AbstractProcessor { @Override public boolean tryProcess() { return tryProcessInternal(null); } InsertWatermarksP(EventTimePolicy<? super T> eventTimePolicy); @Override boolean tryProcess(); @Override boolean saveToSnapshot(); @Override void restoreFromSnapshot(@Nonnull Object key, @Nonnull Object value); @Override boolean finishSnapshotRestore();  }
@Test public void when_idleTimeout_then_idleMessageAfterTimeout() throws Exception { createProcessor(100); resultToCheck.clear(); long start = System.nanoTime(); doAndDrain(() -> p.tryProcess(0, item(10))); assertEquals(asList(wm(10 - LAG), item(10)), resultToCheck); resultToCheck.clear(); long elapsedMs; do { assertTrue(p.tryProcess()); elapsedMs = NANOSECONDS.toMillis(System.nanoTime() - start); outbox.drainQueueAndReset(0, resultToCheck, false); if (elapsedMs < 100) { assertTrue("outbox should be empty, elapsedMs=" + elapsedMs, resultToCheck.isEmpty()); } else if (!resultToCheck.isEmpty()) { System.out.println("WM emitted after " + elapsedMs + "ms (shortly after 100 was expected)"); assertEquals(singletonList(IDLE_MESSAGE), resultToCheck); break; } LockSupport.parkNanos(MILLISECONDS.toNanos(1)); } while (elapsedMs < 1000); }