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

Code before:
            mockTime.sleep(2);
        }

        assertThat((double) iteratorDurationAvgMetric.metricValue(), equalTo(2.0 * TimeUnit.MILLISECONDS.toNanos(1)));
        assertThat((double) iteratorDurationMaxMetric.metricValue(), equalTo(2.0 * TimeUnit.MILLISECONDS.toNanos(1)));

        try (final KeyValueIterator<String, String> iterator = metered.all()) {
            // nothing to do, just close immediately
            mockTime.sleep(3);
        }

        assertThat((double) iteratorDurationAvgMetric.metricValue(), equalTo(2.5 * TimeUnit.MILLISECONDS.toNanos(1)));
        assertThat((double) iteratorDurationMaxMetric.metricValue(), equalTo(3.0 * TimeUnit.MILLISECONDS.toNanos(1)));
    }

    @SuppressWarnings("unused")
    @Test
    public void shouldTrackOldestOpenIteratorTimestamp() {
        setUp();
        when(inner.all()).thenReturn(KeyValueIterators.emptyIterator());
        init();

        final KafkaMetric oldestIteratorTimestampMetric = metric("oldest-iterator-open-since-ms");
        assertThat(oldestIteratorTimestampMetric, not(nullValue()));

        KeyValueIterator<String, String> second = null;
        final long secondTimestamp;
        try {
            try (final KeyValueIterator<String, String> unused = metered.all()) {
                final long oldestTimestamp = mockTime.milliseconds();
                assertThat((Long) oldestIteratorTimestampMetric.metricValue(), equalTo(oldestTimestamp));
                mockTime.sleep(100);

                // open a second iterator before closing the first to test that we still produce the first iterator's timestamp
                second = metered.all();
                secondTimestamp = mockTime.milliseconds();
                assertThat((Long) oldestIteratorTimestampMetric.metricValue(), equalTo(oldestTimestamp));
                mockTime.sleep(100);
            }

            // now that the first iterator is closed, check that the timestamp has advanced to the still open second iterator
            assertThat((Long) oldestIteratorTimestampMetric.metricValue(), equalTo(secondTimestamp));
        } finally {
            if (second != null) {
                second.close();
            }
        }
        // no open iterators left, timestamp should be reset to 0
        assertThat((Long) oldestIteratorTimestampMetric.metricValue(), equalTo(0L));
    }


Review comment (suggestion):
Added this to validate the `KafkaMetric.value()` returns `0L` the window and session tests had this line so I put here to be consistent.

Code after:
            mockTime.sleep(2);
        }

        assertThat((double) iteratorDurationAvgMetric.metricValue(), equalTo(2.0 * TimeUnit.MILLISECONDS.toNanos(1)));
        assertThat((double) iteratorDurationMaxMetric.metricValue(), equalTo(2.0 * TimeUnit.MILLISECONDS.toNanos(1)));

        try (final KeyValueIterator<String, String> iterator = metered.all()) {
            // nothing to do, just close immediately
            mockTime.sleep(3);
        }

        assertThat((double) iteratorDurationAvgMetric.metricValue(), equalTo(2.5 * TimeUnit.MILLISECONDS.toNanos(1)));
        assertThat((double) iteratorDurationMaxMetric.metricValue(), equalTo(3.0 * TimeUnit.MILLISECONDS.toNanos(1)));
    }

    @SuppressWarnings("unused")
    @Test
    public void shouldTrackOldestOpenIteratorTimestamp() {
        setUp();
        when(inner.all()).thenReturn(KeyValueIterators.emptyIterator());
        init();

        final KafkaMetric oldestIteratorTimestampMetric = metric("oldest-iterator-open-since-ms");
        assertThat(oldestIteratorTimestampMetric, not(nullValue()));

        assertThat(oldestIteratorTimestampMetric.metricValue(), equalTo(0L));

        KeyValueIterator<String, String> second = null;
        final long secondTimestamp;
        try {
            try (final KeyValueIterator<String, String> unused = metered.all()) {
                final long oldestTimestamp = mockTime.milliseconds();
                assertThat((Long) oldestIteratorTimestampMetric.metricValue(), equalTo(oldestTimestamp));
                mockTime.sleep(100);

                // open a second iterator before closing the first to test that we still produce the first iterator's timestamp
                second = metered.all();
                secondTimestamp = mockTime.milliseconds();
                assertThat((Long) oldestIteratorTimestampMetric.metricValue(), equalTo(oldestTimestamp));
                mockTime.sleep(100);
            }

            // now that the first iterator is closed, check that the timestamp has advanced to the still open second iterator
            assertThat(oldestIteratorTimestampMetric.metricValue(), equalTo(secondTimestamp));
        } finally {
            if (second != null) {
                second.close();
            }
        }
        // no open iterators left, timestamp should be reset to 0
        assertThat(oldestIteratorTimestampMetric.metricValue(), equalTo(0L));