public void detectAndResolve() {
        createTopologyGraph();
    }
@TestTemplate
    void testWithDynamicFilteringPlan() {
        // no conflicts for dpp pattern
        // 2 --------------------------------------(P1)--- 1 --(P0)--> 0
        //   \                                            /
        //   DynamicFilteringDataCollector               /
        //     \                                        /
        //    DynamicFilteringTableSourceScan --(P0) --/
        TestingBatchExecNode[] nodes = new TestingBatchExecNode[3];
        for (int i = 0; i < nodes.length; i++) {
            nodes[i] = new TestingBatchExecNode("TestingBatchExecNode" + i);
        }

        BatchExecTableSourceScan scan =
                new BatchExecTableSourceScan(
                        new Configuration(),
                        new DynamicTableSourceSpec(null, null),
                        InputProperty.DEFAULT,
                        RowType.of(new IntType(), new IntType(), new IntType()),
                        "DynamicFilteringTableSourceScan");
        BatchExecDynamicFilteringDataCollector collector =
                new BatchExecDynamicFilteringDataCollector(
                        Collections.singletonList(1),
                        new Configuration(),
                        InputProperty.DEFAULT,
                        RowType.of(new IntType()),
                        "DynamicFilteringDataCollector");

        nodes[0].addInput(nodes[1], InputProperty.builder().priority(0).build());
        nodes[1].addInput(nodes[2], InputProperty.builder().priority(1).build());
        nodes[1].addInput(scan, InputProperty.builder().priority(0).build());
        ExecEdge collect2Scan = ExecEdge.builder().source(collector).target(scan).build();
        scan.setInputEdges(Collections.singletonList(collect2Scan));
        ExecEdge toCollector = ExecEdge.builder().source(nodes[2]).target(collector).build();
        collector.setInputEdges(Collections.singletonList(toCollector));

        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[1]),
                        InputProperty.DamBehavior.END_INPUT,
                        streamExchangeMode,
                        configuration);
        resolver.detectAndResolve();

        ExecNode<?> input0 = nodes[1].getInputNodes().get(0);
        ExecNode<?> input1 = nodes[1].getInputNodes().get(1);
        assertThat(input0).isSameAs(nodes[2]);
        assertThat(input1).isSameAs(scan);
    }