private void verifyDbAndCacheAndStatistics(final String segmentName,
                                               final RocksDB db,
                                               final Cache cache,
                                               final Statistics statistics) {
        for (final DbAndCacheAndStatistics valueProviders : storeToValueProviders.values()) {
            verifyConsistencyOfValueProvidersAcrossSegments(segmentName, statistics, valueProviders.statistics, "statistics");
            verifyConsistencyOfValueProvidersAcrossSegments(segmentName, cache, valueProviders.cache, "cache");
            if (db == valueProviders.db) {
                throw new IllegalStateException("DB instance for store " + segmentName + " of task " + taskId +
                    " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
                    "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues");
            }
            if (storeToValueProviders.size() == 1 && cache != valueProviders.cache) {
                singleCache = false;
            } else if (singleCache && cache != valueProviders.cache || !singleCache && cache == valueProviders.cache) {
                throw new IllegalStateException("Caches for store " + storeName + " of task " + taskId +
                    " are either not all distinct or do not all refer to the same cache. This is a bug in Kafka Streams. " +
                    "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues");
            }
        }
    }
The initialisation of the metrics recorder is idempotent.
streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorder.java