SkewReductionPolicy { public boolean observeWm(int queueIndex, final long wmValue) { if (queueWms[queueIndex] >= wmValue) { 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; } SkewReductionPolicy(int numQueues);  SkewReductionPolicy(int numQueues, long maxSkew, long priorityDrainingThreshold, boolean forceAdvanceWm); int toQueueIndex(int drainOrderIdx); boolean observeWm(int queueIndex, final long wmValue); boolean shouldStopDraining(int queueIndex, boolean madeProgress); long bottomObservedWm();  }
@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(); srp.observeWm(0, 5); assertQueuesOrdered(); srp.observeWm(2, 1); assertQueuesOrdered(); for (int i = 0; i < srp.drainOrderToQIdx.length; i++) { srp.observeWm(i, 6); assertQueuesOrdered(); } }