@Override
    public KeyValueSegment getOrCreateSegmentIfLive(final long segmentId,
                                                    final ProcessorContext context,
                                                    final long streamTime) {
        final KeyValueSegment segment = super.getOrCreateSegmentIfLive(segmentId, context, streamTime);
        cleanupExpiredSegments(streamTime);
        return segment;
    }
@Test
    public void shouldCleanupSegmentsThatHaveExpired() {
        final KeyValueSegment segment1 = segments.getOrCreateSegmentIfLive(0, context, -1L);
        final KeyValueSegment segment2 = segments.getOrCreateSegmentIfLive(1, context, -1L);
        final KeyValueSegment segment3 = segments.getOrCreateSegmentIfLive(7, context, SEGMENT_INTERVAL * 7L);
        assertFalse(segment1.isOpen());
        assertFalse(segment2.isOpen());
        assertTrue(segment3.isOpen());
        assertFalse(new File(context.stateDir(), "test/test.0").exists());
        assertFalse(new File(context.stateDir(), "test/test." + SEGMENT_INTERVAL).exists());
        assertTrue(new File(context.stateDir(), "test/test." + 7 * SEGMENT_INTERVAL).exists());
    }