@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 shouldCloseAllOpenSegments() {
        final TimestampedSegment first = segments.getOrCreateSegmentIfLive(0, context, -1L);
        final TimestampedSegment second = segments.getOrCreateSegmentIfLive(1, context, -1L);
        final TimestampedSegment third = segments.getOrCreateSegmentIfLive(2, context, -1L);
        segments.close();

        assertFalse(first.isOpen());
        assertFalse(second.isOpen());
        assertFalse(third.isOpen());
    }