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 testChainedTransformationPropertiesCorrectlySet() {
        ExternalPythonKeyedProcessOperator<?> keyedProcessOperator =
                createKeyedProcessOperator(
                        "f1", new RowTypeInfo(Types.INT(), Types.INT()), Types.STRING());
        ExternalPythonProcessOperator<?, ?> processOperator =
                createProcessOperator("f2", Types.STRING(), Types.STRING());

        Transformation<?> sourceTransformation = mock(SourceTransformation.class);
        OneInputTransformation<?, ?> keyedProcessTransformation =
                new OneInputTransformation(
                        sourceTransformation,
                        "keyedProcess",
                        keyedProcessOperator,
                        keyedProcessOperator.getProducedType(),
                        2);
        keyedProcessTransformation.setUid("uid");
        keyedProcessTransformation.setSlotSharingGroup("group");
        keyedProcessTransformation.setCoLocationGroupKey("col");
        keyedProcessTransformation.setMaxParallelism(64);
        keyedProcessTransformation.declareManagedMemoryUseCaseAtOperatorScope(
                ManagedMemoryUseCase.OPERATOR, 5);
        keyedProcessTransformation.declareManagedMemoryUseCaseAtSlotScope(
                ManagedMemoryUseCase.PYTHON);
        keyedProcessTransformation.declareManagedMemoryUseCaseAtSlotScope(
                ManagedMemoryUseCase.STATE_BACKEND);
        keyedProcessTransformation.setBufferTimeout(1000L);
        keyedProcessTransformation.setChainingStrategy(ChainingStrategy.HEAD);
        Transformation<?> processTransformation =
                new OneInputTransformation(
                        keyedProcessTransformation,
                        "process",
                        processOperator,
                        processOperator.getProducedType(),
                        2);
        processTransformation.setSlotSharingGroup("group");
        processTransformation.declareManagedMemoryUseCaseAtOperatorScope(
                ManagedMemoryUseCase.OPERATOR, 10);
        processTransformation.declareManagedMemoryUseCaseAtSlotScope(ManagedMemoryUseCase.PYTHON);
        processTransformation.setMaxParallelism(64);
        processTransformation.setBufferTimeout(500L);

        List<Transformation<?>> transformations = new ArrayList<>();
        transformations.add(sourceTransformation);
        transformations.add(keyedProcessTransformation);
        transformations.add(processTransformation);

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

        OneInputTransformation<?, ?> chainedTransformation =
                (OneInputTransformation<?, ?>) optimized.get(1);
        assertThat(chainedTransformation.getParallelism()).isEqualTo(2);
        assertThat(sourceTransformation.getOutputType())
                .isEqualTo(chainedTransformation.getInputType());
        assertThat(processOperator.getProducedType())
                .isEqualTo(chainedTransformation.getOutputType());
        assertThat(keyedProcessTransformation.getUid()).isEqualTo(chainedTransformation.getUid());
        assertThat(chainedTransformation.getSlotSharingGroup().get().getName()).isEqualTo("group");
        assertThat(chainedTransformation.getCoLocationGroupKey()).isEqualTo("col");
        assertThat(chainedTransformation.getMaxParallelism()).isEqualTo(64);
        assertThat(chainedTransformation.getBufferTimeout()).isEqualTo(500L);
        assertThat(
                        (int)
                                chainedTransformation
                                        .getManagedMemoryOperatorScopeUseCaseWeights()
                                        .getOrDefault(ManagedMemoryUseCase.OPERATOR, 0))
                .isEqualTo(15);
        assertThat(chainedTransformation.getOperatorFactory().getChainingStrategy())
                .isEqualTo(ChainingStrategy.HEAD);
        assertThat(chainedTransformation.getManagedMemorySlotScopeUseCases())
                .contains(ManagedMemoryUseCase.PYTHON);
        assertThat(chainedTransformation.getManagedMemorySlotScopeUseCases())
                .contains(ManagedMemoryUseCase.STATE_BACKEND);

        OneInputStreamOperator<?, ?> chainedOperator = chainedTransformation.getOperator();
        assertThat(chainedOperator).isInstanceOf(ExternalPythonKeyedProcessOperator.class);
        validateChainedPythonFunctions(
                ((ExternalPythonKeyedProcessOperator<?>) chainedOperator).getPythonFunctionInfo(),
                "f2",
                "f1");
    }