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

        TimeOrderVerifyingConsumer downstreamConsumer = new TimeOrderVerifyingConsumer();

        ApproximateTimeSorter<TimePojo> sorter = new ApproximateTimeSorter<>(
            Duration.ofSeconds(10),
            downstreamConsumer
        );

        sorter.accept(new TimePojo(EPOCH));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(4)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(8)));
        assertEquals(0, downstreamConsumer.timePojos.size());

        //Adding this point should evict the 1st point because it is more than 10 seconds older than the most recent input
        sorter.accept(new TimePojo(EPOCH.plusSeconds(12))); //evict Epoch + 0
        assertEquals(1, downstreamConsumer.timePojos.size());
        sorter.accept(new TimePojo(EPOCH.plusSeconds(16))); //evict Epoch + 4
        sorter.accept(new TimePojo(EPOCH.plusSeconds(20))); //evict Epoch + 8
        assertEquals(3, downstreamConsumer.timePojos.size());

        //none of these "semi-stale" points are auto evicted because are in the "wait for more data" window
        sorter.accept(new TimePojo(EPOCH.plusSeconds(14)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(13)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(12)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(11)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(10)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(9)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(8)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(7)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(6)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(5)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(4)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(3)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(2)));
        sorter.accept(new TimePojo(EPOCH.plusSeconds(1)));
        sorter.accept(new TimePojo(EPOCH));
        assertEquals(3, downstreamConsumer.timePojos.size());

        try {
            sorter.accept(new TimePojo(EPOCH.minusSeconds(1)));
            //if no exception is thrown above then we should fail
            fail("We expect this point the get evicted -- and trigger the TimeOrderVerifyingConsumer");
        } catch (AssertionError ae) {
            //suppress the expected exception
        }
    }