WatermarkCoalescer { public abstract long observeWm(int queueIndex, long wmValue); 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_i1Idle_i2HasWm_then_forwardedImmediately() { assertEquals(Long.MIN_VALUE, wc.observeWm(0, IDLE_MESSAGE.timestamp())); assertEquals(100, wc.observeWm(1, 100)); }
@Test public void when_i1HasWm_i2Idle_then_forwardedImmediately() { assertEquals(Long.MIN_VALUE, wc.observeWm(0, 100)); assertEquals(100, wc.observeWm(1, IDLE_MESSAGE.timestamp())); }
@Test public void when_i1_active_i2_active_then_wmForwardedImmediately() { assertEquals(Long.MIN_VALUE, wc.observeWm(0, 100)); assertEquals(100, wc.observeWm(1, 101)); assertEquals(101, wc.observeWm(0, 101)); }
@Test public void when_i1_active_i2_idle_then_wmForwardedImmediately() { assertEquals(Long.MIN_VALUE, wc.observeWm(0, 100)); assertEquals(100, wc.observeWm(1, IDLE_MESSAGE.timestamp())); }
@Test public void when_i1_idle_i2_active_then_wmForwardedImmediately() { assertEquals(Long.MIN_VALUE, wc.observeWm(0, IDLE_MESSAGE.timestamp())); assertEquals(100, wc.observeWm(1, 100)); }
@Test public void when_i1_idle_i2_idle_then_idleMessageForwardedImmediately() { assertEquals(Long.MIN_VALUE, wc.observeWm(0, IDLE_MESSAGE.timestamp())); assertEquals(IDLE_MESSAGE.timestamp(), wc.observeWm(1, IDLE_MESSAGE.timestamp())); }
@Test public void when_duplicateIdleMessage_then_processed() { assertEquals(Long.MIN_VALUE, wc.observeWm(0, IDLE_MESSAGE.timestamp())); assertEquals(Long.MIN_VALUE, wc.observeWm(0, IDLE_MESSAGE.timestamp())); }
@Test public void when_allIdleAndDuplicateIdleMessage_then_processed() { assertEquals(Long.MIN_VALUE, wc.observeWm(0, IDLE_MESSAGE.timestamp())); assertEquals(IDLE_MESSAGE.timestamp(), wc.observeWm(1, IDLE_MESSAGE.timestamp())); assertEquals(Long.MIN_VALUE, wc.observeWm(0, IDLE_MESSAGE.timestamp())); }
@Test public void when_wmGoesBack_then_error() { assertEquals(Long.MIN_VALUE, wc.observeWm(0, 10)); exception.expectMessage("not monotonically increasing"); assertEquals(Long.MIN_VALUE, wc.observeWm(0, 9)); }