public void record(final long now) {
        logger.debug("Recording metrics for store {}", storeName);
        long bytesWrittenToDatabase = 0;
        long bytesReadFromDatabase = 0;
        long memtableBytesFlushed = 0;
        long memtableHits = 0;
        long memtableMisses = 0;
        long blockCacheDataHits = 0;
        long blockCacheDataMisses = 0;
        long blockCacheIndexHits = 0;
        long blockCacheIndexMisses = 0;
        long blockCacheFilterHits = 0;
        long blockCacheFilterMisses = 0;
        long writeStallDuration = 0;
        long bytesWrittenDuringCompaction = 0;
        long bytesReadDuringCompaction = 0;
        long numberOfOpenFiles = 0;
        long numberOfFileErrors = 0;
        long memtableFlushTimeSum = 0;
        long memtableFlushTimeCount = 0;
        double memtableFlushTimeMin = Double.MAX_VALUE;
        double memtableFlushTimeMax = 0.0;
        long compactionTimeSum = 0;
        long compactionTimeCount = 0;
        double compactionTimeMin = Double.MAX_VALUE;
        double compactionTimeMax = 0.0;
        boolean shouldRecord = true;
        for (final DbAndCacheAndStatistics valueProviders : storeToValueProviders.values()) {
            if (valueProviders.statistics == null) {
                shouldRecord = false;
                break;
            }
            bytesWrittenToDatabase += valueProviders.statistics.getAndResetTickerCount(TickerType.BYTES_WRITTEN);
            bytesReadFromDatabase += valueProviders.statistics.getAndResetTickerCount(TickerType.BYTES_READ);
            memtableBytesFlushed += valueProviders.statistics.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES);
            memtableHits += valueProviders.statistics.getAndResetTickerCount(TickerType.MEMTABLE_HIT);
            memtableMisses += valueProviders.statistics.getAndResetTickerCount(TickerType.MEMTABLE_MISS);
            blockCacheDataHits += valueProviders.statistics.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT);
            blockCacheDataMisses += valueProviders.statistics.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS);
            blockCacheIndexHits += valueProviders.statistics.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT);
            blockCacheIndexMisses += valueProviders.statistics.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS);
            blockCacheFilterHits += valueProviders.statistics.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT);
            blockCacheFilterMisses += valueProviders.statistics.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS);
            writeStallDuration += valueProviders.statistics.getAndResetTickerCount(TickerType.STALL_MICROS);
            bytesWrittenDuringCompaction += valueProviders.statistics.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES);
            bytesReadDuringCompaction += valueProviders.statistics.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES);
            numberOfOpenFiles += valueProviders.statistics.getAndResetTickerCount(TickerType.NO_FILE_OPENS)
                - valueProviders.statistics.getAndResetTickerCount(TickerType.NO_FILE_CLOSES);
            numberOfFileErrors += valueProviders.statistics.getAndResetTickerCount(TickerType.NO_FILE_ERRORS);
            final HistogramData memtableFlushTimeData = valueProviders.statistics.getHistogramData(HistogramType.FLUSH_TIME);
            memtableFlushTimeSum += memtableFlushTimeData.getSum();
            memtableFlushTimeCount += memtableFlushTimeData.getCount();
            memtableFlushTimeMin = Double.min(memtableFlushTimeMin, memtableFlushTimeData.getMin());
            memtableFlushTimeMax = Double.max(memtableFlushTimeMax, memtableFlushTimeData.getMax());
            final HistogramData compactionTimeData = valueProviders.statistics.getHistogramData(HistogramType.COMPACTION_TIME);
            compactionTimeSum += compactionTimeData.getSum();
            compactionTimeCount += compactionTimeData.getCount();
            compactionTimeMin = Double.min(compactionTimeMin, compactionTimeData.getMin());
            compactionTimeMax = Double.max(compactionTimeMax, compactionTimeData.getMax());
        }
        if (shouldRecord) {
            bytesWrittenToDatabaseSensor.record(bytesWrittenToDatabase, now);
            bytesReadFromDatabaseSensor.record(bytesReadFromDatabase, now);
            memtableBytesFlushedSensor.record(memtableBytesFlushed, now);
            memtableHitRatioSensor.record(computeHitRatio(memtableHits, memtableMisses), now);
            memtableAvgFlushTimeSensor.record(computeAvg(memtableFlushTimeSum, memtableFlushTimeCount), now);
            memtableMinFlushTimeSensor.record(memtableFlushTimeMin, now);
            memtableMaxFlushTimeSensor.record(memtableFlushTimeMax, now);
            blockCacheDataHitRatioSensor.record(computeHitRatio(blockCacheDataHits, blockCacheDataMisses), now);
            blockCacheIndexHitRatioSensor.record(computeHitRatio(blockCacheIndexHits, blockCacheIndexMisses), now);
            blockCacheFilterHitRatioSensor.record(computeHitRatio(blockCacheFilterHits, blockCacheFilterMisses), now);
            writeStallDurationSensor.record(writeStallDuration, now);
            bytesWrittenDuringCompactionSensor.record(bytesWrittenDuringCompaction, now);
            bytesReadDuringCompactionSensor.record(bytesReadDuringCompaction, now);
            compactionTimeAvgSensor.record(computeAvg(compactionTimeSum, compactionTimeCount), now);
            compactionTimeMinSensor.record(compactionTimeMin, now);
            compactionTimeMaxSensor.record(compactionTimeMax, now);
            numberOfOpenFilesSensor.record(numberOfOpenFiles, now);
            numberOfFileErrorsSensor.record(numberOfFileErrors, now);
        }
    }
The initialisation of the metrics recorder is idempotent.
streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorder.java