@Override
    public <T_OTHER, OUT> ProcessConfigurableAndNonKeyedPartitionStream<OUT> connectAndProcess(
            KeyedPartitionStream<K, T_OTHER> other,
            TwoInputNonBroadcastStreamProcessFunction<V, T_OTHER, OUT> processFunction) {
        validateStates(
                processFunction.usesStates(),
                new HashSet<>(
                        Collections.singletonList(StateDeclaration.RedistributionMode.IDENTICAL)));

        TypeInformation<OUT> outTypeInfo =
                StreamUtils.getOutputTypeForTwoInputNonBroadcastProcessFunction(
                        processFunction,
                        getType(),
                        ((KeyedPartitionStreamImpl<K, T_OTHER>) other).getType());

        KeyedTwoInputNonBroadcastProcessOperator<K, V, T_OTHER, OUT> processOperator =
                new KeyedTwoInputNonBroadcastProcessOperator<>(processFunction);
        Transformation<OUT> outTransformation =
                StreamUtils.getTwoInputTransformation(
                        "Keyed-TwoInput-Process",
                        this,
                        (KeyedPartitionStreamImpl<K, T_OTHER>) other,
                        outTypeInfo,
                        processOperator);
        environment.addOperator(outTransformation);
        return StreamUtils.wrapWithConfigureHandle(
                new NonKeyedPartitionStreamImpl<>(environment, outTransformation));
    }
@Test
    void testStateErrorWithConnectKeyedStream() throws Exception {
        ExecutionEnvironmentImpl env = StreamTestUtils.getEnv();
        KeyedPartitionStream<Integer, Integer> stream = createKeyedStream(env);

        assertThatThrownBy(
                        () ->
                                stream.connectAndProcess(
                                        createKeyedStream(
                                                env,
                                                new TestingTransformation<>("t2", Types.LONG, 1),
                                                (KeySelector<Long, Integer>) Math::toIntExact),
                                        new StreamTestUtils
                                                .NoOpTwoInputNonBroadcastStreamProcessFunction(
                                                new HashSet<>(
                                                        Collections.singletonList(
                                                                modeIdenticalStateDeclaration)))))
                .isInstanceOf(IllegalRedistributionModeException.class);
    }