@Override
    public <K, T_OTHER, OUT> ProcessConfigurableAndNonKeyedPartitionStream<OUT> connectAndProcess(
            KeyedPartitionStream<K, T_OTHER> other,
            TwoInputBroadcastStreamProcessFunction<T_OTHER, T, OUT> processFunction) {
        // no state redistribution mode check is required here, since all redistribution modes are
        // acceptable

        TypeInformation<OUT> outTypeInfo =
                StreamUtils.getOutputTypeForTwoInputBroadcastProcessFunction(
                        processFunction,
                        ((KeyedPartitionStreamImpl<K, T_OTHER>) other).getType(),
                        getType());
        KeyedTwoInputBroadcastProcessOperator<K, T_OTHER, T, OUT> processOperator =
                new KeyedTwoInputBroadcastProcessOperator<>(processFunction);
        Transformation<OUT> outTransformation =
                StreamUtils.getTwoInputTransformation(
                        "Broadcast-Keyed-TwoInput-Process",
                        (KeyedPartitionStreamImpl<K, T_OTHER>) other,
                        // we should always take the broadcast input as second input.
                        this,
                        outTypeInfo,
                        processOperator);
        environment.addOperator(outTransformation);
        return StreamUtils.wrapWithConfigureHandle(
                new NonKeyedPartitionStreamImpl<>(environment, outTransformation));
    }
@Test
    void testConnectNonKeyedStream() throws Exception {
        ExecutionEnvironmentImpl env = StreamTestUtils.getEnv();
        BroadcastStreamImpl<Integer> stream =
                new BroadcastStreamImpl<>(env, new TestingTransformation<>("t1", Types.INT, 1));
        NonKeyedPartitionStreamImpl<Long> nonKeyedStream =
                new NonKeyedPartitionStreamImpl<>(
                        env, new TestingTransformation<>("t2", Types.LONG, 2));
        stream.connectAndProcess(
                nonKeyedStream, new StreamTestUtils.NoOpTwoInputBroadcastStreamProcessFunction());
        List<Transformation<?>> transformations = env.getTransformations();
        assertThat(transformations).hasSize(1);
        assertProcessType(transformations.get(0), TwoInputTransformation.class, Types.LONG);
    }