public void addValueProviders(final String segmentName,
                                  final RocksDB db,
                                  final Cache cache,
                                  final Statistics statistics) {
        if (storeToValueProviders.isEmpty()) {
            logger.debug("Adding metrics recorder of task {} to metrics recording trigger", taskId);
            streamsMetrics.rocksDBMetricsRecordingTrigger().addMetricsRecorder(this);
        } else if (storeToValueProviders.containsKey(segmentName)) {
            throw new IllegalStateException("Value providers for store " + segmentName + " of task " + taskId +
                " has been already added. This is a bug in Kafka Streams. " +
                "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues");
        }
        verifyDbAndCacheAndStatistics(segmentName, db, cache, statistics);
        logger.debug("Adding value providers for store {} of task {}", segmentName, taskId);
        storeToValueProviders.put(segmentName, new DbAndCacheAndStatistics(db, cache, statistics));
    }
The initialisation of the metrics recorder is idempotent.
streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorder.java