@Override
    public <OUT> ProcessConfigurableAndNonKeyedPartitionStream<OUT> process(
            OneInputStreamProcessFunction<V, OUT> processFunction) {
        validateStates(
                processFunction.usesStates(),
                new HashSet<>(
                        Collections.singletonList(StateDeclaration.RedistributionMode.IDENTICAL)));

        TypeInformation<OUT> outType;
        outType = StreamUtils.getOutputTypeForOneInputProcessFunction(processFunction, getType());

        KeyedProcessOperator<K, V, OUT> operator = new KeyedProcessOperator<>(processFunction);
        Transformation<OUT> transform =
                StreamUtils.getOneInputKeyedTransformation(
                        "KeyedProcess", this, outType, operator, keySelector, keyType);
        environment.addOperator(transform);
        return StreamUtils.wrapWithConfigureHandle(
                new NonKeyedPartitionStreamImpl<>(environment, transform));
    }
@Test
    void testProcess() throws Exception {
        ExecutionEnvironmentImpl env = StreamTestUtils.getEnv();
        KeyedPartitionStream<Integer, Integer> stream = createKeyedStream(env);
        stream.process(new NoOpOneInputStreamProcessFunction());
        KeyedPartitionStream<Integer, Long> resultStream =
                stream.process(new NoOpOneInputStreamProcessFunction(), Math::toIntExact);
        assertThat(resultStream).isInstanceOf(KeyedPartitionStream.class);
        List<Transformation<?>> transformations = env.getTransformations();
        assertThat(transformations).hasSize(2);
        assertProcessType(transformations.get(0), OneInputTransformation.class, Types.LONG);
        assertProcessType(transformations.get(1), OneInputTransformation.class, Types.LONG);
    }