public boolean observeWm(int queueIndex, final long wmValue) {
        if (queueWms[queueIndex] >= wmValue) {
            // this is possible if force-advancing the watermark because we increase
            // the queueWmValue without receiving watermark from that queue
            if (!forceAdvanceWm) {
                throw new JetException("Watermarks not monotonically increasing on queue: " +
                        "last one=" + queueWms[queueIndex] + ", new one=" + wmValue);
            }
            return false;
        }
        boolean didReorder = adjustDrainingOrder(queueIndex, wmValue);
        queueWms[queueIndex] = wmValue;
        forceAdvanceWmIfConfigured();
        return didReorder;
    }
@Test
    public void when_skewedWm_then_drainOrderCorrect() {
        srp = new SkewReductionPolicy(4, 1000, 500, false);

        srp.observeWm(1, 2);
        assertQueuesOrdered();
        srp.observeWm(2, 0);
        assertQueuesOrdered();
        srp.observeWm(3, 3);
        assertQueuesOrdered();
        srp.observeWm(0, 4);
        assertQueuesOrdered();
        // the most advanced becomes even more advanced
        srp.observeWm(0, 5);
        assertQueuesOrdered();
        // the least advanced advances, but still the least advanced
        srp.observeWm(2, 1);
        assertQueuesOrdered();
        // all queue wms become equal
        for (int i = 0; i < srp.drainOrderToQIdx.length; i++) {
            srp.observeWm(i, 6);
            assertQueuesOrdered();
        }
    }