public static List<Transformation<?>> optimize(List<Transformation<?>> transformations) {
        final Map<Transformation<?>, Set<Transformation<?>>> outputMap =
                buildOutputMap(transformations);

        final LinkedHashSet<Transformation<?>> chainedTransformations = new LinkedHashSet<>();
        final Set<Transformation<?>> alreadyTransformed = Sets.newIdentityHashSet();
        final Queue<Transformation<?>> toTransformQueue = Queues.newArrayDeque(transformations);
        while (!toTransformQueue.isEmpty()) {
            final Transformation<?> transformation = toTransformQueue.poll();
            if (!alreadyTransformed.contains(transformation)) {
                alreadyTransformed.add(transformation);

                final ChainInfo chainInfo = chainWithInputIfPossible(transformation, outputMap);
                chainedTransformations.add(chainInfo.newTransformation);
                chainedTransformations.removeAll(chainInfo.oldTransformations);
                alreadyTransformed.addAll(chainInfo.oldTransformations);

                // Add the chained transformation and its inputs to the to-optimize list
                toTransformQueue.add(chainInfo.newTransformation);
                toTransformQueue.addAll(chainInfo.newTransformation.getInputs());
            }
        }
        return new ArrayList<>(chainedTransformations);
    }
@Test
    void testChainingTwoInputOperators() {
        ExternalPythonKeyedCoProcessOperator<?> keyedCoProcessOperator1 =
                createCoKeyedProcessOperator(
                        "f1",
                        new RowTypeInfo(Types.INT(), Types.STRING()),
                        new RowTypeInfo(Types.INT(), Types.INT()),
                        Types.STRING());
        ExternalPythonProcessOperator<?, ?> processOperator1 =
                createProcessOperator(
                        "f2", new RowTypeInfo(Types.INT(), Types.INT()), Types.STRING());
        ExternalPythonProcessOperator<?, ?> processOperator2 =
                createProcessOperator(
                        "f3", new RowTypeInfo(Types.INT(), Types.INT()), Types.LONG());

        ExternalPythonKeyedProcessOperator<?> keyedProcessOperator2 =
                createKeyedProcessOperator(
                        "f4", new RowTypeInfo(Types.INT(), Types.INT()), Types.STRING());
        ExternalPythonProcessOperator<?, ?> processOperator3 =
                createProcessOperator(
                        "f5", new RowTypeInfo(Types.INT(), Types.INT()), Types.STRING());

        Transformation<?> sourceTransformation1 = mock(SourceTransformation.class);
        Transformation<?> sourceTransformation2 = mock(SourceTransformation.class);
        TwoInputTransformation<?, ?, ?> keyedCoProcessTransformation =
                new TwoInputTransformation(
                        sourceTransformation1,
                        sourceTransformation2,
                        "keyedCoProcess",
                        keyedCoProcessOperator1,
                        keyedCoProcessOperator1.getProducedType(),
                        2);
        Transformation<?> processTransformation1 =
                new OneInputTransformation(
                        keyedCoProcessTransformation,
                        "process",
                        processOperator1,
                        processOperator1.getProducedType(),
                        2);
        Transformation<?> processTransformation2 =
                new OneInputTransformation(
                        processTransformation1,
                        "process",
                        processOperator2,
                        processOperator2.getProducedType(),
                        2);

        OneInputTransformation<?, ?> keyedProcessTransformation =
                new OneInputTransformation(
                        processTransformation2,
                        "keyedProcess",
                        keyedProcessOperator2,
                        keyedProcessOperator2.getProducedType(),
                        2);
        Transformation<?> processTransformation3 =
                new OneInputTransformation(
                        keyedProcessTransformation,
                        "process",
                        processOperator3,
                        processOperator3.getProducedType(),
                        2);

        List<Transformation<?>> transformations = new ArrayList<>();
        transformations.add(sourceTransformation1);
        transformations.add(sourceTransformation2);
        transformations.add(keyedCoProcessTransformation);
        transformations.add(processTransformation1);
        transformations.add(processTransformation2);
        transformations.add(keyedProcessTransformation);
        transformations.add(processTransformation3);

        List<Transformation<?>> optimized =
                PythonOperatorChainingOptimizer.optimize(transformations);
        assertThat(optimized).hasSize(4);

        TwoInputTransformation<?, ?, ?> chainedTransformation1 =
                (TwoInputTransformation<?, ?, ?>) optimized.get(2);
        assertThat(sourceTransformation1.getOutputType())
                .isEqualTo(chainedTransformation1.getInputType1());
        assertThat(sourceTransformation2.getOutputType())
                .isEqualTo(chainedTransformation1.getInputType2());
        assertThat(processOperator2.getProducedType())
                .isEqualTo(chainedTransformation1.getOutputType());

        OneInputTransformation<?, ?> chainedTransformation2 =
                (OneInputTransformation<?, ?>) optimized.get(3);
        assertThat(processOperator2.getProducedType())
                .isEqualTo(chainedTransformation2.getInputType());
        assertThat(processOperator3.getProducedType())
                .isEqualTo(chainedTransformation2.getOutputType());

        TwoInputStreamOperator<?, ?, ?> chainedOperator1 = chainedTransformation1.getOperator();
        assertThat(chainedOperator1).isInstanceOf(ExternalPythonKeyedCoProcessOperator.class);
        validateChainedPythonFunctions(
                ((ExternalPythonKeyedCoProcessOperator<?>) chainedOperator1)
                        .getPythonFunctionInfo(),
                "f3",
                "f2",
                "f1");

        OneInputStreamOperator<?, ?> chainedOperator2 = chainedTransformation2.getOperator();
        assertThat(chainedOperator2).isInstanceOf(ExternalPythonKeyedProcessOperator.class);
        validateChainedPythonFunctions(
                ((ExternalPythonKeyedProcessOperator<?>) chainedOperator2).getPythonFunctionInfo(),
                "f5",
                "f4");
    }