public static <K, N, V, S extends State>
            InternalKvState<K, N, ?> createStateAndWrapWithLatencyTrackingIfEnabled(
                    InternalKvState<K, N, ?> kvState,
                    StateDescriptor<S, V> stateDescriptor,
                    LatencyTrackingStateConfig latencyTrackingStateConfig)
                    throws Exception {
        if (latencyTrackingStateConfig.isEnabled()) {
            return new LatencyTrackingStateFactory<>(
                            kvState, stateDescriptor, latencyTrackingStateConfig)
                    .createState();
        }
        return kvState;
    }
@TestTemplate
    @SuppressWarnings("unchecked")
    <K, N> void testTrackAggregatingState() throws Exception {
        InternalAggregatingState<K, N, Long, Long, Long> aggregatingState =
                mock(InternalAggregatingState.class);
        AggregatingStateDescriptor<Long, Long, Long> aggregatingStateDescriptor =
                new AggregatingStateDescriptor<>(
                        "aggregate",
                        new AggregateFunction<Long, Long, Long>() {
                            private static final long serialVersionUID = 1L;

                            @Override
                            public Long createAccumulator() {
                                return 0L;
                            }

                            @Override
                            public Long add(Long value, Long accumulator) {
                                return value + accumulator;
                            }

                            @Override
                            public Long getResult(Long accumulator) {
                                return accumulator;
                            }

                            @Override
                            public Long merge(Long a, Long b) {
                                return a + b;
                            }
                        },
                        Long.class);
        InternalKvState<K, N, ?> latencyTrackingState =
                LatencyTrackingStateFactory.createStateAndWrapWithLatencyTrackingIfEnabled(
                        aggregatingState,
                        aggregatingStateDescriptor,
                        getLatencyTrackingStateConfig());
        if (enableLatencyTracking) {
            assertThat(latencyTrackingState).isInstanceOf(LatencyTrackingAggregatingState.class);
        } else {
            assertThat(latencyTrackingState).isEqualTo(aggregatingState);
        }
    }