class StreamsMetricsImpl implements StreamsMetrics {

    public <T> void addStoreLevelMutableMetric(final String taskId,
                                               final String metricsScope,
                                               final String storeName,
                                               final String name,
                                               final String description,
                                               final RecordingLevel recordingLevel,
                                               final Gauge<T> valueProvider) {
        final MetricName metricName = metrics.metricName(
            name,
            STATE_STORE_LEVEL_GROUP,
            description,
            storeLevelTagMap(taskId, metricsScope, storeName)
        );
        if (metrics.metric(metricName) == null) {
            final MetricConfig metricConfig = new MetricConfig().recordLevel(recordingLevel);
            final String key = storeSensorPrefix(Thread.currentThread().getName(), taskId, storeName);
            metrics.addMetric(metricName, metricConfig, valueProvider);
            storeLevelMetrics.computeIfAbsent(key, ignored -> new LinkedList<>()).push(metricName);
        }
    }

    public  StreamsMetricsImpl(final Metrics metrics,
                              final String clientId,
                              final String builtInMetricsVersion,
                              final Time time);

}