public <T> org.apache.flink.api.common.state.v2.ValueState<T> getValueState(
            org.apache.flink.runtime.state.v2.ValueStateDescriptor<T> stateProperties) {
        KeyedStateStoreV2 keyedStateStoreV2 =
                checkPreconditionsAndGetKeyedStateStoreV2(stateProperties);
        return keyedStateStoreV2.getValueState(stateProperties);
    }
@Test
    void testV2ValueStateInstantiation() throws Exception {

        final ExecutionConfig config = new ExecutionConfig();
        SerializerConfig serializerConfig = config.getSerializerConfig();
        serializerConfig.registerKryoType(Path.class);

        final AtomicReference<Object> descriptorCapture = new AtomicReference<>();

        StreamingRuntimeContext context = createRuntimeContext(descriptorCapture, config);
        org.apache.flink.runtime.state.v2.ValueStateDescriptor<TaskInfo> descr =
                new org.apache.flink.runtime.state.v2.ValueStateDescriptor<>(
                        "name", TypeInformation.of(TaskInfo.class), serializerConfig);
        context.getValueState(descr);

        org.apache.flink.runtime.state.v2.ValueStateDescriptor<?> descrIntercepted =
                (org.apache.flink.runtime.state.v2.ValueStateDescriptor<?>) descriptorCapture.get();
        TypeSerializer<?> serializer = descrIntercepted.getSerializer();

        // check that the Path class is really registered, i.e., the execution config was applied
        assertThat(serializer).isInstanceOf(KryoSerializer.class);
        assertThat(((KryoSerializer<?>) serializer).getKryo().getRegistration(Path.class).getId())
                .isPositive();
    }