TimestampSortFunction implements SortFunction<Long, T> { @Override public Collection<T> onTimer(Long watermark) { List<T> results = new ArrayList<>(); Tuple2<Long,T> head = queue.peek(); while (head != null && head.getF0() <= watermark) { results.add(head.getF1()); queue.remove(head); head = queue.peek(); } return results; } TimestampSortFunction();  TimestampSortFunction(int capacity); @Override void process(Long timestamp, Long watermark, T value); @Override Collection<T> onTimer(Long watermark);  }
@Test public void onTimer() { sortFunction.onTimer(2L); clone.onTimer(2L); }