kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject [kafka] branch trunk updated: KAFKA-8578: Add basic functionality to expose RocksDB metrics (#6979)
Date Fri, 02 Aug 2019 16:51:20 GMT
This is an automated email from the ASF dual-hosted git repository.

guozhang pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new a7d0fdd  KAFKA-8578: Add basic functionality to expose RocksDB metrics (#6979)
a7d0fdd is described below

commit a7d0fdd534ef55533a868ea7388bbc081ee42718
Author: cadonna <bruno@confluent.io>
AuthorDate: Fri Aug 2 18:51:03 2019 +0200

    KAFKA-8578: Add basic functionality to expose RocksDB metrics (#6979)
    
    * Adds RocksDBMetrics class that provides methods to get sensors from the Kafka metrics registry and to setup the sensors to record RocksDB metrics
    
    * Extends StreamsMetricsImpl with functionality to add the required metrics to the sensors.
    
    Reviewers: Boyang Chen <boyang@confluent.io>, Bill Bejeck <bill@confluent.io>, Matthias J. Sax <matthias@confluent.io>, John Roesler <vvcephei@users.noreply.github.com>, Guozhang Wang <wangguoz@gmail.com>
---
 .../internals/metrics/StreamsMetricsImpl.java      |  83 ++++-
 .../state/internals/metrics/RocksDBMetrics.java    | 382 +++++++++++++++++++++
 .../internals/metrics/StreamsMetricsImplTest.java  | 113 +++++-
 .../internals/metrics/RocksDBMetricsTest.java      | 283 +++++++++++++++
 4 files changed, 852 insertions(+), 9 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java
index b6bfcc5..ae3d953 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java
@@ -23,9 +23,12 @@ import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.Sensor.RecordingLevel;
 import org.apache.kafka.common.metrics.stats.Avg;
 import org.apache.kafka.common.metrics.stats.CumulativeCount;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
 import org.apache.kafka.common.metrics.stats.Max;
 import org.apache.kafka.common.metrics.stats.Rate;
+import org.apache.kafka.common.metrics.stats.Value;
 import org.apache.kafka.common.metrics.stats.WindowedCount;
+import org.apache.kafka.common.metrics.stats.WindowedSum;
 import org.apache.kafka.streams.StreamsMetrics;
 
 import java.util.Arrays;
@@ -54,17 +57,21 @@ public class StreamsMetricsImpl implements StreamsMetrics {
 
     public static final String THREAD_ID_TAG = "client-id";
     public static final String TASK_ID_TAG = "task-id";
+    public static final String STORE_ID_TAG = "state-id";
 
     public static final String ALL_TASKS = "all";
 
     public static final String LATENCY_SUFFIX = "-latency";
     public static final String AVG_SUFFIX = "-avg";
     public static final String MAX_SUFFIX = "-max";
+    public static final String MIN_SUFFIX = "-min";
     public static final String RATE_SUFFIX = "-rate";
     public static final String TOTAL_SUFFIX = "-total";
+    public static final String RATIO_SUFFIX = "-ratio";
 
     public static final String THREAD_LEVEL_GROUP = "stream-metrics";
     public static final String TASK_LEVEL_GROUP = "stream-task-metrics";
+    public static final String STATE_LEVEL_GROUP = "stream-state-metrics";
 
     public static final String PROCESSOR_NODE_METRICS_GROUP = "stream-processor-node-metrics";
     public static final String PROCESSOR_NODE_ID_TAG = "processor-node-id";
@@ -123,6 +130,18 @@ public class StreamsMetricsImpl implements StreamsMetrics {
         }
     }
 
+    public Map<String, String> taskLevelTagMap(final String taskName) {
+        final Map<String, String> tagMap = threadLevelTagMap();
+        tagMap.put(TASK_ID_TAG, taskName);
+        return tagMap;
+    }
+
+    public Map<String, String> storeLevelTagMap(final String taskName, final String storeType, final String storeName) {
+        final Map<String, String> tagMap = taskLevelTagMap(taskName);
+        tagMap.put(storeType + "-" + STORE_ID_TAG, storeName);
+        return tagMap;
+    }
+
     public final Sensor taskLevelSensor(final String taskName,
                                         final String sensorName,
                                         final RecordingLevel recordingLevel,
@@ -237,9 +256,7 @@ public class StreamsMetricsImpl implements StreamsMetrics {
             if (!storeLevelSensors.containsKey(key)) {
                 storeLevelSensors.put(key, new LinkedList<>());
             }
-
             final String fullSensorName = key + SENSOR_NAME_DELIMITER + sensorName;
-
             final Sensor sensor = metrics.sensor(fullSensorName, recordingLevel, parents);
 
             storeLevelSensors.get(key).push(fullSensorName);
@@ -454,12 +471,62 @@ public class StreamsMetricsImpl implements StreamsMetrics {
                                                  final String group,
                                                  final Map<String, String> tags,
                                                  final String operation) {
-        addInvocationRateAndCount(sensor,
-                                  group,
-                                  tags,
-                                  operation,
-                                  "The total number of " + operation,
-                                  "The average per-second number of " + operation);
+        addInvocationRateAndCount(
+            sensor,
+            group,
+            tags,
+            operation,
+            "The total number of " + operation,
+            "The average per-second number of " + operation
+        );
+    }
+
+    public static void addRateOfSumAndSumMetricsToSensor(final Sensor sensor,
+                                                         final String group,
+                                                         final Map<String, String> tags,
+                                                         final String operation,
+                                                         final String descriptionOfRate,
+                                                         final String descriptionOfTotal) {
+        addRateOfSumMetricToSensor(sensor, group, tags, operation, descriptionOfRate);
+        addSumMetricToSensor(sensor, group, tags, operation, descriptionOfTotal);
+    }
+
+    public static void addRateOfSumMetricToSensor(final Sensor sensor,
+                                                  final String group,
+                                                  final Map<String, String> tags,
+                                                  final String operation,
+                                                  final String description) {
+        sensor.add(new MetricName(operation + RATE_SUFFIX, group, description, tags),
+                   new Rate(TimeUnit.SECONDS, new WindowedSum()));
+    }
+
+    public static void addSumMetricToSensor(final Sensor sensor,
+                                            final String group,
+                                            final Map<String, String> tags,
+                                            final String operation,
+                                            final String description) {
+        sensor.add(new MetricName(operation + TOTAL_SUFFIX, group, description, tags), new CumulativeSum());
+    }
+
+    public static void addValueMetricToSensor(final Sensor sensor,
+                                              final String group,
+                                              final Map<String, String> tags,
+                                              final String name,
+                                              final String description) {
+        sensor.add(new MetricName(name, group, description, tags), new Value());
+    }
+
+    public static void addAvgAndSumMetricsToSensor(final Sensor sensor,
+                                                   final String group,
+                                                   final Map<String, String> tags,
+                                                   final String metricNamePrefix,
+                                                   final String descriptionOfAvg,
+                                                   final String descriptionOfTotal) {
+        sensor.add(new MetricName(metricNamePrefix + AVG_SUFFIX, group, descriptionOfAvg, tags), new Avg());
+        sensor.add(
+            new MetricName(metricNamePrefix + TOTAL_SUFFIX, group, descriptionOfTotal, tags),
+            new CumulativeSum()
+        );
     }
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetrics.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetrics.java
new file mode 100644
index 0000000..95cb522
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetrics.java
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.state.internals.metrics;
+
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.Sensor.RecordingLevel;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.AVG_SUFFIX;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.MAX_SUFFIX;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.MIN_SUFFIX;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.RATIO_SUFFIX;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.STATE_LEVEL_GROUP;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addRateOfSumMetricToSensor;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addRateOfSumAndSumMetricsToSensor;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgAndSumMetricsToSensor;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addSumMetricToSensor;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addValueMetricToSensor;
+
+public class RocksDBMetrics {
+    private RocksDBMetrics() {}
+
+    private static final String BYTES_WRITTEN_TO_DB = "bytes-written";
+    private static final String BYTES_READ_FROM_DB = "bytes-read";
+    private static final String MEMTABLE_BYTES_FLUSHED = "memtable-bytes-flushed";
+    private static final String MEMTABLE_HIT_RATIO = "memtable-hit" + RATIO_SUFFIX;
+    private static final String MEMTABLE_FLUSH_TIME = "memtable-flush-time";
+    private static final String MEMTABLE_FLUSH_TIME_AVG = MEMTABLE_FLUSH_TIME + AVG_SUFFIX;
+    private static final String MEMTABLE_FLUSH_TIME_MIN = MEMTABLE_FLUSH_TIME + MIN_SUFFIX;
+    private static final String MEMTABLE_FLUSH_TIME_MAX = MEMTABLE_FLUSH_TIME + MAX_SUFFIX;
+    private static final String WRITE_STALL_DURATION = "write-stall-duration";
+    private static final String BLOCK_CACHE_DATA_HIT_RATIO = "block-cache-data-hit" + RATIO_SUFFIX;
+    private static final String BLOCK_CACHE_INDEX_HIT_RATIO = "block-cache-index-hit" + RATIO_SUFFIX;
+    private static final String BLOCK_CACHE_FILTER_HIT_RATIO = "block-cache-filter-hit" + RATIO_SUFFIX;
+    private static final String BYTES_READ_DURING_COMPACTION = "bytes-read-compaction";
+    private static final String BYTES_WRITTEN_DURING_COMPACTION = "bytes-written-compaction";
+    private static final String COMPACTION_TIME = "compaction-time";
+    private static final String COMPACTION_TIME_AVG = COMPACTION_TIME + AVG_SUFFIX;
+    private static final String COMPACTION_TIME_MIN = COMPACTION_TIME + MIN_SUFFIX;
+    private static final String COMPACTION_TIME_MAX = COMPACTION_TIME + MAX_SUFFIX;
+    private static final String NUMBER_OF_OPEN_FILES = "number-open-files";
+    private static final String NUMBER_OF_FILE_ERRORS = "number-file-errors";
+
+    private static final String BYTES_WRITTEN_TO_DB_RATE_DESCRIPTION =
+        "Average number of bytes written per second to the RocksDB state store";
+    private static final String BYTES_WRITTEN_TO_DB_TOTAL_DESCRIPTION =
+        "Total number of bytes written to the RocksDB state store";
+    private static final String BYTES_READ_FROM_DB_RATE_DESCRIPTION =
+        "Average number of bytes read per second from the RocksDB state store";
+    private static final String BYTES_READ_FROM_DB_TOTAL_DESCRIPTION =
+        "Total number of bytes read from the RocksDB state store";
+    private static final String MEMTABLE_BYTES_FLUSHED_RATE_DESCRIPTION =
+        "Average number of bytes flushed per second from the memtable to disk";
+    private static final String MEMTABLE_BYTES_FLUSHED_TOTAL_DESCRIPTION =
+        "Total number of bytes flushed from the memtable to disk";
+    private static final String MEMTABLE_HIT_RATIO_DESCRIPTION =
+        "Ratio of memtable hits relative to all lookups to the memtable";
+    private static final String MEMTABLE_FLUSH_TIME_AVG_DESCRIPTION =
+        "Average time spent on flushing the memtable to disk in ms";
+    private static final String MEMTABLE_FLUSH_TIME_MIN_DESCRIPTION =
+        "Minimum time spent on flushing the memtable to disk in ms";
+    private static final String MEMTABLE_FLUSH_TIME_MAX_DESCRIPTION =
+        "Maximum time spent on flushing the memtable to disk in ms";
+    private static final String WRITE_STALL_DURATION_AVG_DESCRIPTION = "Average duration of write stalls in ms";
+    private static final String WRITE_STALL_DURATION_TOTAL_DESCRIPTION = "Total duration of write stalls in ms";
+    private static final String BLOCK_CACHE_DATA_HIT_RATIO_DESCRIPTION =
+        "Ratio of block cache hits for data relative to all lookups for data to the block cache";
+    private static final String BLOCK_CACHE_INDEX_HIT_RATIO_DESCRIPTION =
+        "Ratio of block cache hits for indexes relative to all lookups for indexes to the block cache";
+    private static final String BLOCK_CACHE_FILTER_HIT_RATIO_DESCRIPTION =
+        "Ratio of block cache hits for filters relative to all lookups for filters to the block cache";
+    private static final String BYTES_READ_DURING_COMPACTION_DESCRIPTION =
+        "Average number of bytes read per second during compaction";
+    private static final String BYTES_WRITTEN_DURING_COMPACTION_DESCRIPTION =
+        "Average number of bytes written per second during compaction";
+    private static final String COMPACTION_TIME_AVG_DESCRIPTION = "Average time spent on compaction in ms";
+    private static final String COMPACTION_TIME_MIN_DESCRIPTION = "Minimum time spent on compaction in ms";
+    private static final String COMPACTION_TIME_MAX_DESCRIPTION = "Maximum time spent on compaction in ms";
+    private static final String NUMBER_OF_OPEN_FILES_DESCRIPTION = "Number of currently open files";
+    private static final String NUMBER_OF_FILE_ERRORS_DESCRIPTION = "Total number of file errors occurred";
+
+    public static class RocksDBMetricContext {
+        private final String taskName;
+        private final String storeType;
+        private final String storeName;
+
+        public RocksDBMetricContext(final String taskName, final String storeType, final String storeName) {
+            this.taskName = taskName;
+            this.storeType = "rocksdb-" + storeType;
+            this.storeName = storeName;
+        }
+
+        public String taskName() {
+            return taskName;
+        }
+        public String storeType() {
+            return storeType;
+        }
+        public String storeName() {
+            return storeName;
+        }
+    }
+
+    public static Sensor bytesWrittenToDatabaseSensor(final StreamsMetricsImpl streamsMetrics,
+                                                      final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, BYTES_WRITTEN_TO_DB);
+        addRateOfSumAndSumMetricsToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            BYTES_WRITTEN_TO_DB,
+            BYTES_WRITTEN_TO_DB_RATE_DESCRIPTION,
+            BYTES_WRITTEN_TO_DB_TOTAL_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor bytesReadFromDatabaseSensor(final StreamsMetricsImpl streamsMetrics,
+                                                     final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, BYTES_READ_FROM_DB);
+        addRateOfSumAndSumMetricsToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            BYTES_READ_FROM_DB,
+            BYTES_READ_FROM_DB_RATE_DESCRIPTION,
+            BYTES_READ_FROM_DB_TOTAL_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor memtableBytesFlushedSensor(final StreamsMetricsImpl streamsMetrics,
+                                                    final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, MEMTABLE_BYTES_FLUSHED);
+        addRateOfSumAndSumMetricsToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            MEMTABLE_BYTES_FLUSHED,
+            MEMTABLE_BYTES_FLUSHED_RATE_DESCRIPTION,
+            MEMTABLE_BYTES_FLUSHED_TOTAL_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor memtableHitRatioSensor(final StreamsMetricsImpl streamsMetrics,
+                                                final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, MEMTABLE_HIT_RATIO);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            MEMTABLE_HIT_RATIO,
+            MEMTABLE_HIT_RATIO_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor memtableAvgFlushTimeSensor(final StreamsMetricsImpl streamsMetrics,
+                                                    final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, MEMTABLE_FLUSH_TIME_AVG);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            MEMTABLE_FLUSH_TIME_AVG,
+            MEMTABLE_FLUSH_TIME_AVG_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor memtableMinFlushTimeSensor(final StreamsMetricsImpl streamsMetrics,
+                                                    final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, MEMTABLE_FLUSH_TIME_MIN);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            MEMTABLE_FLUSH_TIME_MIN,
+            MEMTABLE_FLUSH_TIME_MIN_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor memtableMaxFlushTimeSensor(final StreamsMetricsImpl streamsMetrics,
+                                                    final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, MEMTABLE_FLUSH_TIME_MAX);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            MEMTABLE_FLUSH_TIME_MAX,
+            MEMTABLE_FLUSH_TIME_MAX_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor writeStallDurationSensor(final StreamsMetricsImpl streamsMetrics,
+                                                  final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, WRITE_STALL_DURATION);
+        addAvgAndSumMetricsToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            WRITE_STALL_DURATION,
+            WRITE_STALL_DURATION_AVG_DESCRIPTION,
+            WRITE_STALL_DURATION_TOTAL_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor blockCacheDataHitRatioSensor(final StreamsMetricsImpl streamsMetrics,
+                                                      final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, BLOCK_CACHE_DATA_HIT_RATIO);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            BLOCK_CACHE_DATA_HIT_RATIO,
+            BLOCK_CACHE_DATA_HIT_RATIO_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor blockCacheIndexHitRatioSensor(final StreamsMetricsImpl streamsMetrics,
+                                                       final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, BLOCK_CACHE_INDEX_HIT_RATIO);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics.storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            BLOCK_CACHE_INDEX_HIT_RATIO,
+            BLOCK_CACHE_INDEX_HIT_RATIO_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor blockCacheFilterHitRatioSensor(final StreamsMetricsImpl streamsMetrics,
+                                                        final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, BLOCK_CACHE_FILTER_HIT_RATIO);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            BLOCK_CACHE_FILTER_HIT_RATIO,
+            BLOCK_CACHE_FILTER_HIT_RATIO_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor bytesReadDuringCompactionSensor(final StreamsMetricsImpl streamsMetrics,
+                                                         final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, BYTES_READ_DURING_COMPACTION);
+        addRateOfSumMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            BYTES_READ_DURING_COMPACTION,
+            BYTES_READ_DURING_COMPACTION_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor bytesWrittenDuringCompactionSensor(final StreamsMetricsImpl streamsMetrics,
+                                                            final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, BYTES_WRITTEN_DURING_COMPACTION);
+        addRateOfSumMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            BYTES_WRITTEN_DURING_COMPACTION,
+            BYTES_WRITTEN_DURING_COMPACTION_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor compactionTimeAvgSensor(final StreamsMetricsImpl streamsMetrics,
+                                                 final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, COMPACTION_TIME_AVG);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            COMPACTION_TIME_AVG,
+            COMPACTION_TIME_AVG_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor compactionTimeMinSensor(final StreamsMetricsImpl streamsMetrics,
+                                                 final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, COMPACTION_TIME_MIN);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            COMPACTION_TIME_MIN,
+            COMPACTION_TIME_MIN_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor compactionTimeMaxSensor(final StreamsMetricsImpl streamsMetrics,
+                                                 final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, COMPACTION_TIME_MAX);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            COMPACTION_TIME_MAX,
+            COMPACTION_TIME_MAX_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor numberOfOpenFilesSensor(final StreamsMetricsImpl streamsMetrics,
+                                                 final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, NUMBER_OF_OPEN_FILES);
+        addValueMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            NUMBER_OF_OPEN_FILES,
+            NUMBER_OF_OPEN_FILES_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    public static Sensor numberOfFileErrorsSensor(final StreamsMetricsImpl streamsMetrics,
+                                                  final RocksDBMetricContext metricContext) {
+        final Sensor sensor = createSensor(streamsMetrics, metricContext, NUMBER_OF_FILE_ERRORS);
+        addSumMetricToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            streamsMetrics
+                .storeLevelTagMap(metricContext.taskName(), metricContext.storeType(), metricContext.storeName()),
+            NUMBER_OF_FILE_ERRORS,
+            NUMBER_OF_FILE_ERRORS_DESCRIPTION
+        );
+        return sensor;
+    }
+
+    private static Sensor createSensor(final StreamsMetricsImpl streamsMetrics,
+                                       final RocksDBMetricContext metricContext,
+                                       final String sensorName) {
+        return streamsMetrics.storeLevelSensor(
+            metricContext.taskName(),
+            metricContext.storeName(),
+            sensorName,
+            RecordingLevel.DEBUG);
+    }
+}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImplTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImplTest.java
index f381a58..678d9f3 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImplTest.java
@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.processor.internals.metrics;
 
-
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.metrics.KafkaMetric;
 import org.apache.kafka.common.metrics.MetricConfig;
@@ -28,6 +27,7 @@ import org.easymock.EasyMock;
 import org.easymock.EasyMockSupport;
 import org.junit.Test;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -37,6 +37,8 @@ import static org.apache.kafka.common.utils.Utils.mkMap;
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.PROCESSOR_NODE_METRICS_GROUP;
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgMaxLatency;
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
@@ -49,6 +51,15 @@ public class StreamsMetricsImplTest extends EasyMockSupport {
     private final static String SENSOR_NAME_DELIMITER = ".s.";
     private final static String INTERNAL_PREFIX = "internal";
 
+    private final Metrics metrics = new Metrics();
+    private final Sensor sensor = metrics.sensor("dummy");
+    private final String metricNamePrefix = "metric";
+    private final String group = "group";
+    private final Map<String, String> tags = mkMap(mkEntry("tag", "value"));
+    private final String description1 = "description number one";
+    private final String description2 = "description number two";
+    private final MockTime time = new MockTime(0);
+
     @Test
     public void shouldGetThreadLevelSensor() {
         final Metrics metrics = mock(Metrics.class);
@@ -230,6 +241,106 @@ public class StreamsMetricsImplTest extends EasyMockSupport {
             assertEquals(i, Math.round(totalMetric.measurable().measure(config, time.milliseconds())));
             sensor.record(latency, time.milliseconds());
         }
+    }
+
+    @Test
+    public void shouldGetStoreLevelTagMap() {
+        final String threadName = "test-thread";
+        final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(metrics, threadName);
+        final String taskName = "test-task";
+        final String storeType = "remote-window";
+        final String storeName = "window-keeper";
+
+        final Map<String, String> tagMap = streamsMetrics.storeLevelTagMap(taskName, storeType, storeName);
 
+        assertThat(tagMap.size(), equalTo(3));
+        assertThat(tagMap.get(StreamsMetricsImpl.THREAD_ID_TAG), equalTo(threadName));
+        assertThat(tagMap.get(StreamsMetricsImpl.TASK_ID_TAG), equalTo(taskName));
+        assertThat(tagMap.get(storeType + "-" + StreamsMetricsImpl.STORE_ID_TAG), equalTo(storeName));
+    }
+
+    @Test
+    public void shouldAddAmountRateAndSum() {
+        StreamsMetricsImpl
+            .addRateOfSumAndSumMetricsToSensor(sensor, group, tags, metricNamePrefix, description1, description2);
+
+        final double valueToRecord1 = 18.0;
+        final double valueToRecord2 = 72.0;
+        final long defaultWindowSizeInSeconds = Duration.ofMillis(new MetricConfig().timeWindowMs()).getSeconds();
+        final double expectedRateMetricValue = (valueToRecord1 + valueToRecord2) / defaultWindowSizeInSeconds;
+        verifyMetric(metricNamePrefix + "-rate", description1, valueToRecord1, valueToRecord2, expectedRateMetricValue);
+        final double expectedSumMetricValue = 2 * valueToRecord1 + 2 * valueToRecord2; // values are recorded once for each metric verification
+        verifyMetric(metricNamePrefix + "-total", description2, valueToRecord1, valueToRecord2, expectedSumMetricValue);
+        assertThat(metrics.metrics().size(), equalTo(2 + 1)); // one metric is added automatically in the constructor of Metrics
+    }
+
+    @Test
+    public void shouldAddSum() {
+        StreamsMetricsImpl.addSumMetricToSensor(sensor, group, tags, metricNamePrefix, description1);
+
+        final double valueToRecord1 = 18.0;
+        final double valueToRecord2 = 42.0;
+        final double expectedSumMetricValue = valueToRecord1 + valueToRecord2;
+        verifyMetric(metricNamePrefix + "-total", description1, valueToRecord1, valueToRecord2, expectedSumMetricValue);
+        assertThat(metrics.metrics().size(), equalTo(1 + 1)); // one metric is added automatically in the constructor of Metrics
+    }
+
+    @Test
+    public void shouldAddAmountRate() {
+        StreamsMetricsImpl.addRateOfSumMetricToSensor(sensor, group, tags, metricNamePrefix, description1);
+
+        final double valueToRecord1 = 18.0;
+        final double valueToRecord2 = 72.0;
+        final long defaultWindowSizeInSeconds = Duration.ofMillis(new MetricConfig().timeWindowMs()).getSeconds();
+        final double expectedRateMetricValue = (valueToRecord1 + valueToRecord2) / defaultWindowSizeInSeconds;
+        verifyMetric(metricNamePrefix + "-rate", description1, valueToRecord1, valueToRecord2, expectedRateMetricValue);
+        assertThat(metrics.metrics().size(), equalTo(1 + 1)); // one metric is added automatically in the constructor of Metrics
+    }
+
+    @Test
+    public void shouldAddValue() {
+        StreamsMetricsImpl.addValueMetricToSensor(sensor, group, tags, metricNamePrefix, description1);
+
+        final KafkaMetric ratioMetric = metrics.metric(new MetricName(metricNamePrefix, group, description1, tags));
+        assertThat(ratioMetric, is(notNullValue()));
+        final MetricConfig metricConfig = new MetricConfig();
+        final double value1 = 42.0;
+        sensor.record(value1);
+        assertThat(ratioMetric.measurable().measure(metricConfig, time.milliseconds()), equalTo(42.0));
+        final double value2 = 18.0;
+        sensor.record(value2);
+        assertThat(ratioMetric.measurable().measure(metricConfig, time.milliseconds()), equalTo(18.0));
+        assertThat(metrics.metrics().size(), equalTo(1 + 1)); // one metric is added automatically in the constructor of Metrics
+    }
+
+    @Test
+    public void shouldAddAvgAndTotalMetricsToSensor() {
+        StreamsMetricsImpl
+            .addAvgAndSumMetricsToSensor(sensor, group, tags, metricNamePrefix, description1, description2);
+
+        final double valueToRecord1 = 18.0;
+        final double valueToRecord2 = 42.0;
+        final double expectedAvgMetricValue = (valueToRecord1 + valueToRecord2) / 2;
+        verifyMetric(metricNamePrefix + "-avg", description1, valueToRecord1, valueToRecord2, expectedAvgMetricValue);
+        final double expectedSumMetricValue = 2 * valueToRecord1 + 2 * valueToRecord2; // values are recorded once for each metric verification
+        verifyMetric(metricNamePrefix + "-total", description2, valueToRecord1, valueToRecord2, expectedSumMetricValue);
+        assertThat(metrics.metrics().size(), equalTo(2 + 1)); // one metric is added automatically in the constructor of Metrics
+    }
+
+    private void verifyMetric(final String name,
+                              final String description,
+                              final double valueToRecord1,
+                              final double valueToRecord2,
+                              final double expectedMetricValue) {
+        final KafkaMetric metric = metrics
+            .metric(new MetricName(name, group, description, tags));
+        assertThat(metric, is(notNullValue()));
+        assertThat(metric.metricName().description(), equalTo(description));
+        sensor.record(valueToRecord1, time.milliseconds());
+        sensor.record(valueToRecord2, time.milliseconds());
+        assertThat(
+            metric.measurable().measure(new MetricConfig(), time.milliseconds()),
+            equalTo(expectedMetricValue)
+        );
     }
 }
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsTest.java
new file mode 100644
index 0000000..6c8b9d8
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsTest.java
@@ -0,0 +1,283 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.state.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.Sensor.RecordingLevel;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.internals.metrics.RocksDBMetrics.RocksDBMetricContext;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.easymock.EasyMock.expect;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.powermock.api.easymock.PowerMock.createStrictMock;
+import static org.powermock.api.easymock.PowerMock.mockStatic;
+import static org.powermock.api.easymock.PowerMock.replay;
+import static org.powermock.api.easymock.PowerMock.replayAll;
+import static org.powermock.api.easymock.PowerMock.verify;
+import static org.powermock.api.easymock.PowerMock.verifyAll;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(StreamsMetricsImpl.class)
+public class RocksDBMetricsTest {
+
+    private static final String STATE_LEVEL_GROUP = "stream-state-metrics";
+    private static final String STORE_TYPE_PREFIX = "rocksdb-";
+    private final String taskName = "task";
+    private final String storeType = "storeType";
+    private final String storeName = "store";
+    private final Metrics metrics = new Metrics();
+    private final Sensor sensor = metrics.sensor("dummy");
+    private final StreamsMetricsImpl streamsMetrics = createStrictMock(StreamsMetricsImpl.class);
+    private final Map<String, String> tags = Collections.singletonMap("hello", "world");
+
+    private interface SensorCreator {
+        Sensor sensor(final StreamsMetricsImpl streamsMetrics, final RocksDBMetricContext metricContext);
+    }
+
+    @Test
+    public void shouldGetBytesWrittenSensor() {
+        final String metricNamePrefix = "bytes-written";
+        final String descriptionOfTotal = "Total number of bytes written to the RocksDB state store";
+        final String descriptionOfRate = "Average number of bytes written per second to the RocksDB state store";
+        verifyRateAndTotalSensor(
+            metricNamePrefix,
+            descriptionOfTotal,
+            descriptionOfRate,
+            RocksDBMetrics::bytesWrittenToDatabaseSensor
+        );
+    }
+
+    @Test
+    public void shouldGetBytesReadSensor() {
+        final String metricNamePrefix = "bytes-read";
+        final String descriptionOfTotal = "Total number of bytes read from the RocksDB state store";
+        final String descriptionOfRate = "Average number of bytes read per second from the RocksDB state store";
+        verifyRateAndTotalSensor(
+            metricNamePrefix,
+            descriptionOfTotal,
+            descriptionOfRate,
+            RocksDBMetrics::bytesReadFromDatabaseSensor
+        );
+    }
+
+    @Test
+    public void shouldGetMemtableHitRatioSensor() {
+        final String metricNamePrefix = "memtable-hit-ratio";
+        final String description = "Ratio of memtable hits relative to all lookups to the memtable";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::memtableHitRatioSensor);
+    }
+
+    @Test
+    public void shouldGetMemtableBytesFlushedSensor() {
+        final String metricNamePrefix = "memtable-bytes-flushed";
+        final String descriptionOfTotal = "Total number of bytes flushed from the memtable to disk";
+        final String descriptionOfRate = "Average number of bytes flushed per second from the memtable to disk";
+        verifyRateAndTotalSensor(
+            metricNamePrefix,
+            descriptionOfTotal,
+            descriptionOfRate,
+            RocksDBMetrics::memtableBytesFlushedSensor
+        );
+    }
+
+    @Test
+    public void shouldGetMemtableAvgFlushTimeSensor() {
+        final String metricNamePrefix = "memtable-flush-time-avg";
+        final String description = "Average time spent on flushing the memtable to disk in ms";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::memtableAvgFlushTimeSensor);
+    }
+
+    @Test
+    public void shouldGetMemtableMinFlushTimeSensor() {
+        final String metricNamePrefix = "memtable-flush-time-min";
+        final String description = "Minimum time spent on flushing the memtable to disk in ms";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::memtableMinFlushTimeSensor);
+    }
+
+    @Test
+    public void shouldGetMemtableMaxFlushTimeSensor() {
+        final String metricNamePrefix = "memtable-flush-time-max";
+        final String description = "Maximum time spent on flushing the memtable to disk in ms";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::memtableMaxFlushTimeSensor);
+    }
+
+    @Test
+    public void shouldGetWriteStallDurationSensor() {
+        final String metricNamePrefix = "write-stall-duration";
+        final String descriptionOfAvg = "Average duration of write stalls in ms";
+        final String descriptionOfTotal = "Total duration of write stalls in ms";
+        setupStreamsMetricsMock(metricNamePrefix);
+        StreamsMetricsImpl.addAvgAndSumMetricsToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            tags,
+            metricNamePrefix,
+            descriptionOfAvg,
+            descriptionOfTotal
+        );
+
+        replayCallAndVerify(RocksDBMetrics::writeStallDurationSensor);
+    }
+
+    @Test
+    public void shouldGetBlockCacheDataHitRatioSensor() {
+        final String metricNamePrefix = "block-cache-data-hit-ratio";
+        final String description =
+            "Ratio of block cache hits for data relative to all lookups for data to the block cache";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::blockCacheDataHitRatioSensor);
+    }
+
+    @Test
+    public void shouldGetBlockCacheIndexHitRatioSensor() {
+        final String metricNamePrefix = "block-cache-index-hit-ratio";
+        final String description =
+            "Ratio of block cache hits for indexes relative to all lookups for indexes to the block cache";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::blockCacheIndexHitRatioSensor);
+    }
+
+    @Test
+    public void shouldGetBlockCacheFilterHitRatioSensor() {
+        final String metricNamePrefix = "block-cache-filter-hit-ratio";
+        final String description =
+            "Ratio of block cache hits for filters relative to all lookups for filters to the block cache";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::blockCacheFilterHitRatioSensor);
+    }
+
+    @Test
+    public void shouldGetBytesReadDuringCompactionSensor() {
+        final String metricNamePrefix = "bytes-read-compaction";
+        final String description = "Average number of bytes read per second during compaction";
+        verifyRateSensor(metricNamePrefix, description, RocksDBMetrics::bytesReadDuringCompactionSensor);
+    }
+
+    @Test
+    public void shouldGetBytesWrittenDuringCompactionSensor() {
+        final String metricNamePrefix = "bytes-written-compaction";
+        final String description = "Average number of bytes written per second during compaction";
+        verifyRateSensor(metricNamePrefix, description, RocksDBMetrics::bytesWrittenDuringCompactionSensor);
+    }
+
+    @Test
+    public void shouldGetCompactionTimeAvgSensor() {
+        final String metricNamePrefix = "compaction-time-avg";
+        final String description = "Average time spent on compaction in ms";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::compactionTimeAvgSensor);
+    }
+
+    @Test
+    public void shouldGetCompactionTimeMinSensor() {
+        final String metricNamePrefix = "compaction-time-min";
+        final String description = "Minimum time spent on compaction in ms";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::compactionTimeMinSensor);
+    }
+
+    @Test
+    public void shouldGetCompactionTimeMaxSensor() {
+        final String metricNamePrefix = "compaction-time-max";
+        final String description = "Maximum time spent on compaction in ms";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::compactionTimeMaxSensor);
+    }
+
+    @Test
+    public void shouldGetNumberOfOpenFilesSensor() {
+        final String metricNamePrefix = "number-open-files";
+        final String description = "Number of currently open files";
+        verifyValueSensor(metricNamePrefix, description, RocksDBMetrics::numberOfOpenFilesSensor);
+    }
+
+    @Test
+    public void shouldGetNumberOfFilesErrors() {
+        final String metricNamePrefix = "number-file-errors";
+        final String description = "Total number of file errors occurred";
+        setupStreamsMetricsMock(metricNamePrefix);
+        StreamsMetricsImpl.addSumMetricToSensor(sensor, STATE_LEVEL_GROUP, tags, metricNamePrefix, description);
+
+        replayCallAndVerify(RocksDBMetrics::numberOfFileErrorsSensor);
+    }
+
+    private void verifyRateAndTotalSensor(final String metricNamePrefix,
+                                          final String descriptionOfTotal,
+                                          final String descriptionOfRate,
+                                          final SensorCreator sensorCreator) {
+        setupStreamsMetricsMock(metricNamePrefix);
+        StreamsMetricsImpl.addRateOfSumAndSumMetricsToSensor(
+            sensor,
+            STATE_LEVEL_GROUP,
+            tags,
+            metricNamePrefix,
+            descriptionOfRate,
+            descriptionOfTotal
+        );
+
+        replayCallAndVerify(sensorCreator);
+    }
+
+    private void verifyRateSensor(final String metricNamePrefix,
+                                  final String description,
+                                  final SensorCreator sensorCreator) {
+        setupStreamsMetricsMock(metricNamePrefix);
+        StreamsMetricsImpl.addRateOfSumMetricToSensor(sensor, STATE_LEVEL_GROUP, tags, metricNamePrefix, description);
+
+        replayCallAndVerify(sensorCreator);
+    }
+
+    private void verifyValueSensor(final String metricNamePrefix,
+                                   final String description,
+                                   final SensorCreator sensorCreator) {
+        setupStreamsMetricsMock(metricNamePrefix);
+        StreamsMetricsImpl.addValueMetricToSensor(sensor, STATE_LEVEL_GROUP, tags, metricNamePrefix, description);
+
+        replayCallAndVerify(sensorCreator);
+    }
+
+    private void setupStreamsMetricsMock(final String metricNamePrefix) {
+        mockStatic(StreamsMetricsImpl.class);
+        expect(streamsMetrics.storeLevelSensor(
+            taskName,
+            storeName,
+            metricNamePrefix,
+            RecordingLevel.DEBUG
+        )).andReturn(sensor);
+        expect(streamsMetrics.storeLevelTagMap(
+            taskName,
+            STORE_TYPE_PREFIX + storeType,
+            storeName
+        )).andReturn(tags);
+    }
+
+    private void replayCallAndVerify(final SensorCreator sensorCreator) {
+        replayAll();
+        replay(StreamsMetricsImpl.class);
+
+        final Sensor sensor =
+            sensorCreator.sensor(streamsMetrics, new RocksDBMetricContext(taskName, storeType, storeName));
+
+        verifyAll();
+        verify(StreamsMetricsImpl.class);
+
+        assertThat(sensor, is(this.sensor));
+    }
+}


Mime
View raw message