public void detectAndResolve() {
        createTopologyGraph();
    }
@TestTemplate
    void testDetectAndResolve() {
        // P = InputProperty.DamBehavior.PIPELINED, E = InputProperty.DamBehavior.END_INPUT
        // P100 = PIPELINED + priority 100
        //
        // 0 --------(P0)----> 1 --(P0)-----------> 7
        //  \                    \-(P0)-> 2 -(P0)--/
        //   \-------(P0)----> 3 --(P1)-----------/
        //    \------(P0)----> 4 --(P10)---------/
        //     \              /                 /
        //      \    8 -(P0)-<                 /
        //       \            \               /
        //        \--(E0)----> 5 --(P10)-----/
        // 6 ---------(P100)----------------/
        TestingBatchExecNode[] nodes = new TestingBatchExecNode[9];
        for (int i = 0; i < nodes.length; i++) {
            nodes[i] = new TestingBatchExecNode("TestingBatchExecNode" + i);
        }
        nodes[1].addInput(nodes[0], InputProperty.builder().priority(0).build());
        nodes[2].addInput(nodes[1], InputProperty.builder().priority(0).build());
        nodes[3].addInput(nodes[0], InputProperty.builder().priority(0).build());
        nodes[4].addInput(nodes[8], InputProperty.builder().priority(0).build());
        nodes[4].addInput(nodes[0], InputProperty.builder().priority(0).build());
        nodes[5].addInput(nodes[8], InputProperty.builder().priority(0).build());
        nodes[5].addInput(
                nodes[0],
                InputProperty.builder()
                        .damBehavior(InputProperty.DamBehavior.END_INPUT)
                        .priority(0)
                        .build());
        nodes[7].addInput(nodes[1], InputProperty.builder().priority(0).build());
        nodes[7].addInput(nodes[2], InputProperty.builder().priority(0).build());
        nodes[7].addInput(nodes[3], InputProperty.builder().priority(1).build());
        nodes[7].addInput(nodes[4], InputProperty.builder().priority(10).build());
        nodes[7].addInput(nodes[5], InputProperty.builder().priority(10).build());
        nodes[7].addInput(nodes[6], InputProperty.builder().priority(100).build());

        Configuration configuration = new Configuration();
        BatchShuffleMode batchShuffleMode = batchShuffleModeAndStreamExchangeMode.f0;
        StreamExchangeMode streamExchangeMode = batchShuffleModeAndStreamExchangeMode.f1;
        configuration.set(ExecutionOptions.BATCH_SHUFFLE_MODE, batchShuffleMode);

        InputPriorityConflictResolver resolver =
                new InputPriorityConflictResolver(
                        Collections.singletonList(nodes[7]),
                        InputProperty.DamBehavior.END_INPUT,
                        streamExchangeMode,
                        configuration);
        resolver.detectAndResolve();
        assertThat(nodes[7].getInputNodes().get(0)).isEqualTo(nodes[1]);
        assertThat(nodes[7].getInputNodes().get(1)).isEqualTo(nodes[2]);
        assertThat(nodes[7].getInputNodes().get(2)).isInstanceOf(BatchExecExchange.class);
        assertThat(((BatchExecExchange) nodes[7].getInputNodes().get(2)).getRequiredExchangeMode())
                .isEqualTo(Optional.of(streamExchangeMode));
        assertThat(nodes[7].getInputNodes().get(2).getInputEdges().get(0).getSource())
                .isEqualTo(nodes[3]);
        assertThat(nodes[7].getInputNodes().get(3)).isInstanceOf(BatchExecExchange.class);
        assertThat(((BatchExecExchange) nodes[7].getInputNodes().get(3)).getRequiredExchangeMode())
                .isEqualTo(Optional.of(streamExchangeMode));
        assertThat(nodes[7].getInputNodes().get(3).getInputEdges().get(0).getSource())
                .isEqualTo(nodes[4]);
        assertThat(nodes[7].getInputNodes().get(4)).isEqualTo(nodes[5]);
        assertThat(nodes[7].getInputNodes().get(5)).isEqualTo(nodes[6]);
    }