@Override
    public void openExisting(final ProcessorContext context, final long streamTime) {
        metricsRecorder.init(ProcessorContextUtils.getMetricsImpl(context), context.taskId());
        super.openExisting(context, streamTime);
    }
Manages the {@link KeyValueSegment}s that are used by the {@link RocksDBSegmentedBytesStore}
streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueSegments.java