LatencyTrackingAggregatingState(
            String stateName,
            InternalAggregatingState<K, N, IN, ACC, OUT> original,
            LatencyTrackingStateConfig latencyTrackingStateConfig) {
        super(
                original,
                new AggregatingStateLatencyMetrics(
                        stateName,
                        latencyTrackingStateConfig.getMetricGroup(),
                        latencyTrackingStateConfig.getSampleInterval(),
                        latencyTrackingStateConfig.getHistorySize(),
                        latencyTrackingStateConfig.isStateNameAsVariable()));
    }
@Test
    @SuppressWarnings({"unchecked", "rawtypes"})
    void testLatencyTrackingAggregatingState() throws Exception {
        AbstractKeyedStateBackend<Integer> keyedBackend = createKeyedBackend(getKeySerializer());
        try {
            LatencyTrackingAggregatingState<Integer, VoidNamespace, Long, Long, Long>
                    latencyTrackingState =
                            (LatencyTrackingAggregatingState)
                                    createLatencyTrackingState(keyedBackend, getStateDescriptor());
            latencyTrackingState.setCurrentNamespace(VoidNamespace.INSTANCE);
            LatencyTrackingAggregatingState.AggregatingStateLatencyMetrics
                    latencyTrackingStateMetric =
                            latencyTrackingState.getLatencyTrackingStateMetric();

            assertThat(latencyTrackingStateMetric.getAddCount()).isZero();
            assertThat(latencyTrackingStateMetric.getGetCount()).isZero();
            assertThat(latencyTrackingStateMetric.getMergeNamespaceCount()).isZero();

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

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

                latencyTrackingState.mergeNamespaces(
                        VoidNamespace.INSTANCE, Collections.emptyList());
                assertThat(latencyTrackingStateMetric.getMergeNamespaceCount())
                        .isEqualTo(expectedResult);
            }
        } finally {
            if (keyedBackend != null) {
                keyedBackend.close();
                keyedBackend.dispose();
            }
        }
    }