LatencyTrackingMapState(
            String stateName,
            InternalMapState<K, N, UK, UV> original,
            LatencyTrackingStateConfig latencyTrackingStateConfig) {
        super(
                original,
                new MapStateLatencyMetrics(
                        stateName,
                        latencyTrackingStateConfig.getMetricGroup(),
                        latencyTrackingStateConfig.getSampleInterval(),
                        latencyTrackingStateConfig.getHistorySize(),
                        latencyTrackingStateConfig.isStateNameAsVariable()));
    }
@Test
    @SuppressWarnings({"unchecked", "rawtypes"})
    void testLatencyTrackingMapState() throws Exception {
        AbstractKeyedStateBackend<Integer> keyedBackend = createKeyedBackend(getKeySerializer());
        try {
            LatencyTrackingMapState<Integer, VoidNamespace, Long, Double> latencyTrackingState =
                    (LatencyTrackingMapState)
                            createLatencyTrackingState(keyedBackend, getStateDescriptor());
            latencyTrackingState.setCurrentNamespace(VoidNamespace.INSTANCE);
            LatencyTrackingMapState.MapStateLatencyMetrics latencyTrackingStateMetric =
                    latencyTrackingState.getLatencyTrackingStateMetric();

            assertThat(latencyTrackingStateMetric.getContainsCount()).isZero();
            assertThat(latencyTrackingStateMetric.getEntriesInitCount()).isZero();
            assertThat(latencyTrackingStateMetric.getGetCount()).isZero();
            assertThat(latencyTrackingStateMetric.getIsEmptyCount()).isZero();
            assertThat(latencyTrackingStateMetric.getIteratorInitCount()).isZero();
            assertThat(latencyTrackingStateMetric.getIteratorHasNextCount()).isZero();
            assertThat(latencyTrackingStateMetric.getIteratorNextCount()).isZero();
            assertThat(latencyTrackingStateMetric.getKeysInitCount()).isZero();
            assertThat(latencyTrackingStateMetric.getValuesInitCount()).isZero();
            assertThat(latencyTrackingStateMetric.getIteratorRemoveCount()).isZero();
            assertThat(latencyTrackingStateMetric.getPutAllCount()).isZero();
            assertThat(latencyTrackingStateMetric.getPutCount()).isZero();
            assertThat(latencyTrackingStateMetric.getRemoveCount()).isZero();

            setCurrentKey(keyedBackend);
            ThreadLocalRandom random = ThreadLocalRandom.current();
            for (int index = 1; index <= SAMPLE_INTERVAL; index++) {
                int expectedResult = index == SAMPLE_INTERVAL ? 0 : index;
                latencyTrackingState.put(random.nextLong(), random.nextDouble());
                assertThat(latencyTrackingStateMetric.getPutCount()).isEqualTo(expectedResult);

                latencyTrackingState.putAll(
                        Collections.singletonMap(random.nextLong(), random.nextDouble()));
                assertThat(latencyTrackingStateMetric.getPutAllCount()).isEqualTo(expectedResult);

                latencyTrackingState.get(random.nextLong());
                assertThat(latencyTrackingStateMetric.getGetCount()).isEqualTo(expectedResult);

                latencyTrackingState.remove(random.nextLong());
                assertThat(latencyTrackingStateMetric.getRemoveCount()).isEqualTo(expectedResult);

                latencyTrackingState.contains(random.nextLong());
                assertThat(latencyTrackingStateMetric.getContainsCount()).isEqualTo(expectedResult);

                latencyTrackingState.isEmpty();
                assertThat(latencyTrackingStateMetric.getIsEmptyCount()).isEqualTo(expectedResult);

                latencyTrackingState.entries();
                assertThat(latencyTrackingStateMetric.getEntriesInitCount())
                        .isEqualTo(expectedResult);

                latencyTrackingState.keys();
                assertThat(latencyTrackingStateMetric.getKeysInitCount()).isEqualTo(expectedResult);

                latencyTrackingState.values();
                assertThat(latencyTrackingStateMetric.getValuesInitCount())
                        .isEqualTo(expectedResult);

                latencyTrackingState.iterator();
                assertThat(latencyTrackingStateMetric.getIteratorInitCount())
                        .isEqualTo(expectedResult);
            }
        } finally {
            if (keyedBackend != null) {
                keyedBackend.close();
                keyedBackend.dispose();
            }
        }
    }