private Gauge<BigInteger> gaugeToComputeBlockCacheMetrics(final String propertyName) {
        return (metricsConfig, now) -> {
            BigInteger result = BigInteger.valueOf(0);
            for (final DbAndCacheAndStatistics valueProvider : storeToValueProviders.values()) {
                try {
                    if (singleCache) {
                        // values of RocksDB properties are of type unsigned long in C++, i.e., in Java we need to use
                        // BigInteger and construct the object from the byte representation of the value
                        result = new BigInteger(1, longToBytes(
                            valueProvider.db.getLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName)
                        ));
                        break;
                    } else {
                        // values of RocksDB properties are of type unsigned long in C++, i.e., in Java we need to use
                        // BigInteger and construct the object from the byte representation of the value
                        result = result.add(new BigInteger(1, longToBytes(
                            valueProvider.db.getLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName)
                        )));
                    }
                } catch (final RocksDBException e) {
                    throw new ProcessorStateException("Error recording RocksDB metric " + propertyName, e);
                }
            }
            return result;
        };
    }
The initialisation of the metrics recorder is idempotent.
streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorder.java