@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 testOverflowBug() {
        /*
         * It should not be possible to crash an ApproximateTimeSorter by overloading the memory by
         * continuously providing older and older points.
         */

        Duration maxLag = Duration.ofSeconds(3);  //a very small sorting window

        ConsumingArrayList<TimePojo> downstreamConsumer = newConsumingArrayList();

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

        //add Points to the sorter that keep getting older and older
        int NUM_POINTS = 50_000;
        for (int i = 0; i < NUM_POINTS; i++) {
            Instant time = EPOCH.minusSeconds(i);
            sorter.accept(new TimePojo(time));
        }
        int numPointsPastSorter = downstreamConsumer.size();

        /*
         * We want most of the points to pass this sorter because this stream of "perfectly out of
         * order" data should not: (1) lead to an OutOfMemoryException or (2) be a fixable input
         * type. In other words, this data is so bad it isn't reasonable to get it organized.
         */
        assertTrue(numPointsPastSorter >= NUM_POINTS * .99);
    }