@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 shouldNotCreateSegmentThatIsAlreadyExpired() {
        final long streamTime = updateStreamTimeAndCreateSegment(7);
        assertNull(segments.getOrCreateSegmentIfLive(0, context, streamTime));
        assertFalse(new File(context.stateDir(), "test/test.0").exists());
    }