public void detectAndResolve() {
        createTopologyGraph();
    }
@TestTemplate
    void testDeadlockCausedByExchange() {
        // P1 = PIPELINED + priority 1
        //
        // 0 -(P0)-> exchange --(P0)-> 1
        //                    \-(P1)-/
        TestingBatchExecNode[] nodes = new TestingBatchExecNode[2];
        for (int i = 0; i < nodes.length; i++) {
            nodes[i] = new TestingBatchExecNode("TestingBatchExecNode" + i);
        }

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

        BatchExecExchange exchange =
                new BatchExecExchange(
                        configuration,
                        InputProperty.builder()
                                .requiredDistribution(InputProperty.ANY_DISTRIBUTION)
                                .build(),
                        (RowType) nodes[0].getOutputType(),
                        "Exchange");
        exchange.setRequiredExchangeMode(streamExchangeMode);
        ExecEdge execEdge = ExecEdge.builder().source(nodes[0]).target(exchange).build();
        exchange.setInputEdges(Collections.singletonList(execEdge));

        nodes[1].addInput(exchange, InputProperty.builder().priority(0).build());
        nodes[1].addInput(exchange, InputProperty.builder().priority(1).build());

        InputPriorityConflictResolver resolver =
                new InputPriorityConflictResolver(
                        Collections.singletonList(nodes[1]),
                        InputProperty.DamBehavior.END_INPUT,
                        streamExchangeMode,
                        configuration);
        resolver.detectAndResolve();

        ExecNode<?> input0 = nodes[1].getInputNodes().get(0);
        ExecNode<?> input1 = nodes[1].getInputNodes().get(1);
        assertThat(input1).isNotSameAs(input0);

        Consumer<ExecNode<?>> checkExchange =
                execNode -> {
                    assertThat(execNode).isInstanceOf(BatchExecExchange.class);
                    BatchExecExchange e = (BatchExecExchange) execNode;
                    assertThat(e.getRequiredExchangeMode())
                            .isEqualTo(Optional.of(streamExchangeMode));
                    assertThat(e.getInputEdges().get(0).getSource()).isEqualTo(nodes[0]);
                };
        checkExchange.accept(input0);
        checkExchange.accept(input1);
    }