@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 testEmittedPointsAreNotProperlySorted() {

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

        TimeOrderVerifyingConsumer consumer = new TimeOrderVerifyingConsumer();

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

        sorter.accept(new TimePojo(EPOCH));
        assertEquals(0, consumer.size());

        sorter.accept(new TimePojo(EPOCH.plusSeconds(1)));
        assertEquals(0, consumer.size());

        sorter.accept(new TimePojo(EPOCH.plusSeconds(10)));
        assertEquals(2, consumer.size());
        assertEquals(consumer.timeFor(0), EPOCH);
        assertEquals(consumer.timeFor(1), EPOCH.plusSeconds(1));

        sorter.accept(new TimePojo(EPOCH.plusSeconds(5)));
        assertEquals(3, consumer.size());
        assertEquals(consumer.timeFor(2), EPOCH.plusSeconds(5));

        //this TestConsumer fails here because this point is passed through the sorter (after the 5 point was)
        assertThrows(
            AssertionError.class,
            () -> sorter.accept(new TimePojo(EPOCH.plusSeconds(4)))
        );
    }