private void verifyConsistencyOfValueProvidersAcrossSegments(final String segmentName,
                                                                 final Object newValueProvider,
                                                                 final Object oldValueProvider,
                                                                 final String valueProviderName) {
        if (newValueProvider == null && oldValueProvider != null ||
            newValueProvider != null && oldValueProvider == null) {

            final char capitalizedFirstChar = valueProviderName.toUpperCase(Locale.US).charAt(0);
            final StringBuilder capitalizedValueProviderName = new StringBuilder(valueProviderName);
            capitalizedValueProviderName.setCharAt(0, capitalizedFirstChar);
            throw new IllegalStateException(capitalizedValueProviderName +
                " for segment " + segmentName + " of task " + taskId +
                " is" + (newValueProvider == null ? " " : " not ") + "null although the " + valueProviderName +
                " of another segment in this metrics recorder is" + (newValueProvider != null ? " " : " not ") + "null. " +
                "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