@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 shouldCreateSegments() {
        final TimestampedSegment segment1 = segments.getOrCreateSegmentIfLive(0, context, -1L);
        final TimestampedSegment segment2 = segments.getOrCreateSegmentIfLive(1, context, -1L);
        final TimestampedSegment segment3 = segments.getOrCreateSegmentIfLive(2, context, -1L);
        assertTrue(new File(context.stateDir(), "test/test.0").isDirectory());
        assertTrue(new File(context.stateDir(), "test/test." + SEGMENT_INTERVAL).isDirectory());
        assertTrue(new File(context.stateDir(), "test/test." + 2 * SEGMENT_INTERVAL).isDirectory());
        assertTrue(segment1.isOpen());
        assertTrue(segment2.isOpen());
        assertTrue(segment3.isOpen());
    }