WatermarkCoalescer { public abstract long queueDone(int queueIndex); private  WatermarkCoalescer(); abstract long queueDone(int queueIndex); abstract void observeEvent(int queueIndex); abstract long observeWm(int queueIndex, long wmValue); abstract long checkWmHistory(); abstract long coalescedWm(); abstract long topObservedWm(); static WatermarkCoalescer create(int queueCount); static final Watermark IDLE_MESSAGE; }
@Test public void when_allDone_then_noMaxValueEmitted() { assertEquals(Long.MIN_VALUE, wc.queueDone(0)); assertEquals(Long.MIN_VALUE, wc.queueDone(1)); }
@Test public void when_duplicateDoneCall_then_error() { assertEquals(Long.MIN_VALUE, wc.queueDone(0)); exception.expectMessage("Duplicate"); assertEquals(Long.MIN_VALUE, wc.queueDone(0)); }