public List<Watermark> observeWm(int queueIndex, Watermark watermark) {
        if (watermark.equals(IDLE_MESSAGE)) {
            idleQueues[queueIndex] = true;
            boolean allIdle = true;
            // we need to track idle queues for the case when there's no coalescer yet
            for (int i = 0; i < idleQueues.length; i++) {
                if (!doneQueues.contains(i)) {
                    allIdle &= idleQueues[i];
                }
            }

            List<Watermark> watermarks = new ArrayList<>();
            for (Entry<Byte, WatermarkCoalescer> coalescerEntry : coalescers.entrySet()) {
                long observedWm = coalescerEntry.getValue().observeWm(queueIndex, watermark.timestamp());
                assert observedWm != IDLE_MESSAGE_TIME;
                if (observedWm != NO_NEW_WM) {
                    watermarks.add(new Watermark(observedWm, coalescerEntry.getKey()));
                }
                assert coalescerEntry.getValue().idleMessagePending() == allIdle;
            }

            if (allIdle) {
                watermarks.add(IDLE_MESSAGE);
            }
            return watermarks;
        }

        idleQueues[queueIndex] = false;

        WatermarkCoalescer c = coalescer(watermark.key());
        long newWmValue = c.observeWm(queueIndex, watermark.timestamp());
        assert !c.idleMessagePending();
        if (newWmValue == NO_NEW_WM) {
            return emptyList();
        }
        return singletonList(new Watermark(newWmValue, watermark.key()));
    }
@Test
    public void test_initialScenario4() {
        kwc.observeWm(0, wm(42));
        assertEquals(emptyList(), kwc.observeWm(0, IDLE_MESSAGE));
        assertEquals(asList(wm(42), IDLE_MESSAGE), kwc.observeWm(1, IDLE_MESSAGE));
    }