@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;
    }
Manages the {@link KeyValueSegment}s that are used by the {@link RocksDBSegmentedBytesStore}
streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueSegments.java