@Override
    public TimestampedSegment getOrCreateSegment(final long segmentId,
                                                 final ProcessorContext context) {
        if (segments.containsKey(segmentId)) {
            return segments.get(segmentId);
        } else {
            final TimestampedSegment newSegment =
                new TimestampedSegment(segmentName(segmentId), name, segmentId, position, metricsRecorder);

            if (segments.put(segmentId, newSegment) != null) {
                throw new IllegalStateException("TimestampedSegment already exists. Possible concurrent access.");
            }

            newSegment.openDB(context.appConfigs(), context.stateDir());
            return newSegment;
        }
    }
Manages the {@link TimestampedSegment}s that are used by the {@link RocksDBTimestampedSegmentedBytesStore}
streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedSegments.java