Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java (Java)

Code before:
        setUp();
        init();
        final JmxReporter reporter = new JmxReporter();
        final MetricsContext metricsContext = new KafkaMetricsContext("kafka.streams");
        reporter.contextChange(metricsContext);

        metrics.addReporter(reporter);
        assertTrue(reporter.containsMbean(String.format(
            "kafka.streams:type=%s,%s=%s,task-id=%s,%s-state-id=%s",
            STORE_LEVEL_GROUP,
            THREAD_ID_TAG_KEY,
            threadId,
            taskId,
            STORE_TYPE,
            STORE_NAME
        )));
    }

    @Test
    public void shouldRecordRestoreLatencyOnRecordRestoreTime() {
        setUp();
        doNothing().when(inner).init(context, metered);

        init();

        final long restoreTimeNs = 1000;
        metered.recordRestoreTime(restoreTimeNs);

        // it suffices to verify one restore metric since all restore metrics are recorded by the same sensor
        // and the sensor is tested elsewhere
        final KafkaMetric metric = metric("restore-rate");
        assertThat((Double) metric.metricValue(), greaterThan(0.0));
    }

    @Test
    public void shouldWriteBytesToInnerStoreAndRecordPutMetric() {
        setUp();
        doNothing().when(inner).put(KEY_BYTES, VALUE_BYTES);
        init();

        metered.put(KEY, VALUE);

        final KafkaMetric metric = metric("put-rate");
        assertTrue((Double) metric.metricValue() > 0);
    }

    @Test
    public void shouldGetBytesFromInnerStoreAndReturnGetMetric() {
        setUp();
        when(inner.get(KEY_BYTES)).thenReturn(VALUE_BYTES);
        init();

Review comment (suggestion):
```suggestion
        final long restoreTimeNs = 1000L;
```

Code after:
        init();
        final JmxReporter reporter = new JmxReporter();
        final MetricsContext metricsContext = new KafkaMetricsContext("kafka.streams");
        reporter.contextChange(metricsContext);

        metrics.addReporter(reporter);
        assertTrue(reporter.containsMbean(String.format(
            "kafka.streams:type=%s,%s=%s,task-id=%s,%s-state-id=%s",
            STORE_LEVEL_GROUP,
            THREAD_ID_TAG_KEY,
            threadId,
            taskId,
            STORE_TYPE,
            STORE_NAME
        )));
    }

    @Test
    public void shouldRecordRestoreLatencyOnRecordRestoreTime() {
        setUp();
        doNothing().when(inner).init(context, metered);

        init();

        final long restoreTimeNs = 1000L;
        metered.recordRestoreTime(restoreTimeNs);

        // it suffices to verify one restore metric since all restore metrics are recorded by the same sensor
        // and the sensor is tested elsewhere
        final KafkaMetric metric = metric("restore-latency-max");
        assertThat((Double) metric.metricValue(), equalTo((double) restoreTimeNs));
    }

    @Test
    public void shouldWriteBytesToInnerStoreAndRecordPutMetric() {
        setUp();
        doNothing().when(inner).put(KEY_BYTES, VALUE_BYTES);
        init();

        metered.put(KEY, VALUE);

        final KafkaMetric metric = metric("put-rate");
        assertTrue((Double) metric.metricValue() > 0);
    }

    @Test
    public void shouldGetBytesFromInnerStoreAndReturnGetMetric() {
        setUp();
        when(inner.get(KEY_BYTES)).thenReturn(VALUE_BYTES);
        init();