@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 shouldClearSegmentsOnClose() {
        segments.getOrCreateSegmentIfLive(0, context, -1L);
        segments.close();
        assertThat(segments.getSegmentForTimestamp(0), is(nullValue()));
    }