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

        TypeInformation<OUT> outType =
                StreamUtils.getOutputTypeForOneInputProcessFunction(processFunction, getType());
        ProcessOperator<T, OUT> operator = new ProcessOperator<>(processFunction);
        return StreamUtils.wrapWithConfigureHandle(transform("Global Process", outType, operator));
    }
@Test
    void testStateErrorWithOneInputStream() throws Exception {
        ExecutionEnvironmentImpl env = StreamTestUtils.getEnv();
        GlobalStreamImpl<Integer> stream =
                new GlobalStreamImpl<>(env, new TestingTransformation<>("t1", Types.INT, 1));

        assertThatThrownBy(
                        () ->
                                stream.process(
                                        new NoOpOneInputStreamProcessFunction(
                                                new HashSet<>(
                                                        Collections.singletonList(
                                                                modeIdenticalStateDeclaration)))))
                .isInstanceOf(IllegalRedistributionModeException.class);
    }