@Override
    public void accept(T t) {
        updateTimeHighWaterMark(t.time());

        shortTermStorage.add(t);

        drainDueToLatestInput(t); //standard drain policy
        drainDueToTimeHighWaterMark();  //prevent blow-up when data goes backwards in time

        sizeHighWaterMark = Math.max(sizeHighWaterMark, shortTermStorage.size());
    }
@Test
    public void testEmittedPointsAreProperlySorted() {

        Duration maxLag = Duration.ofSeconds(10L); //with this lag things get sorted correctly

        TimeOrderVerifyingConsumer consumer = new TimeOrderVerifyingConsumer();

        ApproximateTimeSorter<TimePojo> sorter = new ApproximateTimeSorter<>(
            maxLag,
            consumer
        );

        for (TimePojo timeBox : testData()) {
            sorter.accept(timeBox);
        }

        assertEquals(7, consumer.size());
    }