@Override
    public TimestampedSegment getOrCreateSegmentIfLive(final long segmentId,
                                                    final ProcessorContext context,
                                                    final long streamTime) {
        final TimestampedSegment segment = super.getOrCreateSegmentIfLive(segmentId, context, streamTime);
        cleanupExpiredSegments(streamTime);
        return segment;
    }
@Test
    public void futureEventsShouldNotCauseSegmentRoll() {
        updateStreamTimeAndCreateSegment(0);
        verifyCorrectSegments(0, 1);
        updateStreamTimeAndCreateSegment(1);
        verifyCorrectSegments(0, 2);
        updateStreamTimeAndCreateSegment(2);
        verifyCorrectSegments(0, 3);
        updateStreamTimeAndCreateSegment(3);
        verifyCorrectSegments(0, 4);
        final long streamTime = updateStreamTimeAndCreateSegment(4);
        verifyCorrectSegments(0, 5);
        segments.getOrCreateSegmentIfLive(5, context, streamTime);
        verifyCorrectSegments(0, 6);
        segments.getOrCreateSegmentIfLive(6, context, streamTime);
        verifyCorrectSegments(0, 7);
    }