TimestampSortFunction implements SortFunction<Long, T> { @Override public void process(Long timestamp, Long watermark, T value) { if (queue.peek() == null || timestamp > watermark) { queue.add(new Tuple2<>(timestamp,value)); } else { logger.warn("Value with old timestamp is discarded {} for {}", timestamp, value); } } TimestampSortFunction();  TimestampSortFunction(int capacity); @Override void process(Long timestamp, Long watermark, T value); @Override Collection<T> onTimer(Long watermark);  }
@Test public void process() { sortFunction.process(1L, 2L, "test1"); clone.process(1L, 2L, "test1"); }