@Override
    public <OUT> ProcessConfigurableAndNonKeyedPartitionStream<OUT> process(
            OneInputStreamProcessFunction<T, OUT> processFunction) {
        validateStates(
                processFunction.usesStates(),
                new HashSet<>(
                        Arrays.asList(
                                StateDeclaration.RedistributionMode.NONE,
                                StateDeclaration.RedistributionMode.IDENTICAL)));

        TypeInformation<OUT> outType =
                StreamUtils.getOutputTypeForOneInputProcessFunction(processFunction, getType());
        ProcessOperator<T, OUT> operator = new ProcessOperator<>(processFunction);
        OneInputTransformation<T, OUT> outputTransform =
                StreamUtils.getOneInputTransformation("Process", this, outType, operator);
        environment.addOperator(outputTransform);
        return StreamUtils.wrapWithConfigureHandle(
                new NonKeyedPartitionStreamImpl<>(environment, outputTransform));
    }
@Test
    void testProcess() throws Exception {
        ExecutionEnvironmentImpl env = StreamTestUtils.getEnv();
        NonKeyedPartitionStreamImpl<Integer> stream =
                new NonKeyedPartitionStreamImpl<>(
                        env, new TestingTransformation<>("t1", Types.INT, 1));
        stream.process(new StreamTestUtils.NoOpOneInputStreamProcessFunction());
        List<Transformation<?>> transformations = env.getTransformations();
        assertThat(transformations).hasSize(1);
        assertProcessType(transformations.get(0), OneInputTransformation.class, Types.LONG);
    }