@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 testAllPointsWithinWindow() {
        /*
         * Confirm that no points are emitted when all the points occur within the time window
         */
        Duration maxLag = Duration.ofMinutes(5);

        ApproximateTimeSorter<TimePojo> sorter = new ApproximateTimeSorter<>(
            maxLag,
            (TimePojo t) -> {
                throw new UnsupportedOperationException("No Point should be forward to this Consumer");
            }
        );

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