@Override
    public Sensor addLatencyRateTotalSensor(final String scopeName,
                                            final String entityName,
                                            final String operationName,
                                            final Sensor.RecordingLevel recordingLevel,
                                            final String... tags) {
        final String threadId = Thread.currentThread().getName();
        final String group = groupNameFromScope(scopeName);
        final Map<String, String> tagMap = customizedTags(threadId, scopeName, entityName, tags);
        final Sensor sensor =
            customInvocationRateAndCountSensor(threadId, group, entityName, operationName, tagMap, recordingLevel);
        addAvgAndMaxToSensor(
            sensor,
            group,
            tagMap,
            operationName + LATENCY_SUFFIX,
            AVG_LATENCY_DESCRIPTION + operationName,
            MAX_LATENCY_DESCRIPTION + operationName
        );

        return sensor;
    }
@Test
    public void testTotalMetricDoesntDecrease() {
        final MockTime time = new MockTime(1);
        final MetricConfig config = new MetricConfig().timeWindow(1, TimeUnit.MILLISECONDS);
        final Metrics metrics = new Metrics(config, time);
        final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(metrics, "", VERSION, time);

        final String scope = "scope";
        final String entity = "entity";
        final String operation = "op";

        final Sensor sensor = streamsMetrics.addLatencyRateTotalSensor(
            scope,
            entity,
            operation,
            RecordingLevel.INFO
        );

        final double latency = 100.0;
        final MetricName totalMetricName = metrics.metricName(
            "op-total",
            "stream-scope-metrics",
            "",
            "thread-id",
            Thread.currentThread().getName(),
            "scope-id",
            "entity"
        );

        final KafkaMetric totalMetric = metrics.metric(totalMetricName);

        for (int i = 0; i < 10; i++) {
            assertEquals(i, Math.round(totalMetric.measurable().measure(config, time.milliseconds())));
            sensor.record(latency, time.milliseconds());
        }
    }