@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 testConnectBroadcastStream() throws Exception {
        ExecutionEnvironmentImpl env = StreamTestUtils.getEnv();
        KeyedPartitionStream<Long, Long> stream =
                createKeyedStream(env, new TestingTransformation<>("t1", Types.LONG, 1), x -> x);
        BroadcastStreamImpl<Integer> s =
                new BroadcastStreamImpl<>(env, new TestingTransformation<>("t2", Types.INT, 1));
        stream.connectAndProcess(
                s, new StreamTestUtils.NoOpTwoInputBroadcastStreamProcessFunction());
        stream.connectAndProcess(
                s, new StreamTestUtils.NoOpTwoInputBroadcastStreamProcessFunction(), x -> x);
        List<Transformation<?>> transformations = env.getTransformations();
        assertThat(transformations).hasSize(2);
        assertProcessType(transformations.get(0), TwoInputTransformation.class, Types.LONG);
        assertProcessType(transformations.get(1), TwoInputTransformation.class, Types.LONG);
    }