public static <IN1, IN2, OUT> TwoInputTransformation<IN1, IN2, OUT> getTwoInputTransformation(
            String operatorName,
            AbstractDataStream<IN1> inputStream1,
            AbstractDataStream<IN2> inputStream2,
            TypeInformation<OUT> outTypeInformation,
            TwoInputStreamOperator<IN1, IN2, OUT> operator) {
        TwoInputTransformation<IN1, IN2, OUT> transform =
                new TwoInputTransformation<>(
                        inputStream1.getTransformation(),
                        inputStream2.getTransformation(),
                        operatorName,
                        SimpleOperatorFactory.of(operator),
                        outTypeInformation,
                        // inputStream1 & 2 share the same env.
                        inputStream1.getEnvironment().getParallelism(),
                        false);

        TypeInformation<?> keyType = null;
        if (inputStream1 instanceof KeyedPartitionStreamImpl) {
            KeyedPartitionStreamImpl<?, IN1> keyedInput1 =
                    (KeyedPartitionStreamImpl<?, IN1>) inputStream1;

            keyType = keyedInput1.getKeyType();

            transform.setStateKeySelectors(keyedInput1.getKeySelector(), null);
            transform.setStateKeyType(keyType);
        }
        if (inputStream2 instanceof KeyedPartitionStreamImpl) {
            KeyedPartitionStreamImpl<?, IN2> keyedInput2 =
                    (KeyedPartitionStreamImpl<?, IN2>) inputStream2;

            TypeInformation<?> keyType2 = keyedInput2.getKeyType();

            if (keyType != null && !(keyType.canEqual(keyType2) && keyType.equals(keyType2))) {
                throw new UnsupportedOperationException(
                        "Key types if input KeyedStreams "
                                + "don't match: "
                                + keyType
                                + " and "
                                + keyType2
                                + ".");
            }

            transform.setStateKeySelectors(
                    transform.getStateKeySelector1(), keyedInput2.getKeySelector());

            // we might be overwriting the one that's already set, but it's the same
            transform.setStateKeyType(keyType2);
        }

        return transform;
    }
@Test
    void testGetTwoInputTransformation() throws Exception {
        ExecutionEnvironmentImpl env = StreamTestUtils.getEnv();
        TwoInputNonBroadcastProcessOperator<Integer, Long, Long> operator =
                new TwoInputNonBroadcastProcessOperator<>(
                        new StreamTestUtils.NoOpTwoInputNonBroadcastStreamProcessFunction());
        TwoInputTransformation<Integer, Long, Long> transformation =
                StreamUtils.getTwoInputTransformation(
                        "op",
                        new NonKeyedPartitionStreamImpl<>(
                                env, new TestingTransformation<>("t1", Types.INT, 1)),
                        new NonKeyedPartitionStreamImpl<>(
                                env, new TestingTransformation<>("t2", Types.LONG, 1)),
                        Types.LONG,
                        operator);
        assertThat(transformation.getOperator()).isEqualTo(operator);
        assertThat(transformation.getOutputType()).isEqualTo(Types.LONG);
        assertThat(transformation.getStateKeySelector1()).isNull();
        assertThat(transformation.getStateKeySelector2()).isNull();
    }