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

Code before:
        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,
            context.taskId().toString(),
            STORE_TYPE,
            STORE_NAME
        )));
    }

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

        store.init(context, store);

        final long restoreTimeNs = 1000;
        store.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 shouldRecordRestoreLatencyOnInit() {
        doNothing().when(innerStoreMock).init(context, store);
        store.init(context, store);

        // 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 shouldPutToInnerStoreAndRecordPutMetrics() {
        final byte[] bytes = "a".getBytes();
        doNothing().when(innerStoreMock).put(eq(Bytes.wrap(bytes)), any(), eq(context.recordContext().timestamp()));

        store.init(context, store);
        store.put("a", "a", context.recordContext().timestamp());

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

Review comment (performance):
With the old code, it was not clear what latency we would measure, to we use `greaterThan`, but know we know that we should record exactly `1000`. Should we use `is(1000L)` instead?

Seems the existing tests for the other store use `assertTrue(...)` should we also update them to use `assertThat` ?

Code after:
        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,
            context.taskId().toString(),
            STORE_TYPE,
            STORE_NAME
        )));
    }

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

        store.init(context, store);

        final long restoreTimeNs = 1000L;
        store.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 shouldPutToInnerStoreAndRecordPutMetrics() {
        final byte[] bytes = "a".getBytes();
        doNothing().when(innerStoreMock).put(eq(Bytes.wrap(bytes)), any(), eq(context.recordContext().timestamp()));

        store.init(context, store);
        store.put("a", "a", context.recordContext().timestamp());

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

    @Test
    public void shouldFetchFromInnerStoreAndRecordFetchMetrics() {
        when(innerStoreMock.fetch(Bytes.wrap("a".getBytes()), 1, 1))
            .thenReturn(KeyValueIterators.emptyWindowStoreIterator());

        store.init(context, store);
        store.fetch("a", ofEpochMilli(1), ofEpochMilli(1)).close(); // recorded on close;

        // it suffices to verify one fetch metric since all fetch metrics are recorded by the same sensor