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 testTrackReducingState() throws Exception {
        InternalReducingState<K, N, Long> reducingState = mock(InternalReducingState.class);
        ReducingStateDescriptor<Long> reducingStateDescriptor =
                new ReducingStateDescriptor<>("reducing", Long::sum, Long.class);
        InternalKvState<K, N, ?> latencyTrackingState =
                LatencyTrackingStateFactory.createStateAndWrapWithLatencyTrackingIfEnabled(
                        reducingState, reducingStateDescriptor, getLatencyTrackingStateConfig());
        if (enableLatencyTracking) {
            assertThat(latencyTrackingState).isInstanceOf(LatencyTrackingReducingState.class);
        } else {
            assertThat(latencyTrackingState).isEqualTo(reducingState);
        }
    }