public static Map<String, List<DataNode>> buildDataNodesMap(final ShardingSphereDatabase database, final Collection<String> tableNames) {
        Optional<ShardingRule> shardingRule = database.getRuleMetaData().findSingleRule(ShardingRule.class);
        Optional<SingleRule> singleRule = database.getRuleMetaData().findSingleRule(SingleRule.class);
        Optional<BroadcastRule> broadcastRule = database.getRuleMetaData().findSingleRule(BroadcastRule.class);
        Map<String, List<DataNode>> result = new HashMap<>(tableNames.size(), 1F);
        // TODO support virtual data source name
        for (String each : tableNames) {
            if (singleRule.isPresent() && singleRule.get().getAttributes().getAttribute(DataNodeRuleAttribute.class).getAllDataNodes().containsKey(each)) {
                result.put(each, new ArrayList<>(singleRule.get().getAttributes().getAttribute(DataNodeRuleAttribute.class).getAllDataNodes().get(each)));
                continue;
            }
            if (shardingRule.isPresent() && shardingRule.get().findShardingTable(each).isPresent()) {
                ShardingTable shardingTable = shardingRule.get().getShardingTable(each);
                result.put(each, shardingTable.getActualDataNodes());
                continue;
            }
            if (broadcastRule.isPresent() && broadcastRule.get().getAttributes().getAttribute(DataNodeRuleAttribute.class).findFirstActualTable(each).isPresent()) {
                result.put(each, Collections.singletonList(broadcastRule.get().getAttributes().getAttribute(DataNodeRuleAttribute.class).getAllDataNodes().get(each).iterator().next()));
                continue;
            }
            throw new PipelineInvalidParameterException(String.format("Not find actual data nodes of `%s`", each));
        }
        return result;
    }
@Test
    void assertBuildDataNodesMap() {
        ShardingSphereDatabase mockDatabase = mock(ShardingSphereDatabase.class);
        RuleMetaData mockRuleMetaData = mock(RuleMetaData.class);
        ShardingRule mockShardingRule = mock(ShardingRule.class);
        ShardingTable mockShardingTable = mock(ShardingTable.class);
        when(mockShardingTable.getActualDataNodes()).thenReturn(Collections.singletonList(new DataNode("ds_0.t_order")));
        when(mockShardingRule.findShardingTable("t_order")).thenReturn(Optional.of(mockShardingTable));
        when(mockShardingRule.getShardingTable("t_order")).thenReturn(mockShardingTable);
        when(mockRuleMetaData.findSingleRule(ShardingRule.class)).thenReturn(Optional.of(mockShardingRule));
        SingleRule singleRule = mock(SingleRule.class);
        when(mockRuleMetaData.findSingleRule(SingleRule.class)).thenReturn(Optional.of(singleRule));
        DataNodeRuleAttribute singleDataNodeRuleAttribute = mock(DataNodeRuleAttribute.class);
        when(singleDataNodeRuleAttribute.getAllDataNodes()).thenReturn(Collections.singletonMap("t_order_item", Collections.singletonList(new DataNode("single.t_order_item"))));
        when(singleRule.getAttributes()).thenReturn(new RuleAttributes(singleDataNodeRuleAttribute));
        when(mockDatabase.getRuleMetaData()).thenReturn(mockRuleMetaData);
        BroadcastRule broadcastRule = mock(BroadcastRule.class, RETURNS_DEEP_STUBS);
        when(mockRuleMetaData.findSingleRule(BroadcastRule.class)).thenReturn(Optional.of(broadcastRule));
        DataNodeRuleAttribute broadcastDataNodeRuleAttribute = mock(DataNodeRuleAttribute.class);
        when(broadcastDataNodeRuleAttribute.findFirstActualTable("t_address")).thenReturn(Optional.of("broadcast.t_address"));
        when(broadcastDataNodeRuleAttribute.getAllDataNodes()).thenReturn(Collections.singletonMap("t_address", Collections.singletonList(new DataNode("broadcast.t_address"))));
        when(broadcastRule.getAttributes()).thenReturn(new RuleAttributes(broadcastDataNodeRuleAttribute));
        Map<String, List<DataNode>> actual = CDCDataNodeUtils.buildDataNodesMap(mockDatabase, Arrays.asList("t_order", "t_order_item", "t_address"));
        assertTrue(actual.containsKey("t_order"));
        assertTrue(actual.containsKey("t_order_item"));
        assertTrue(actual.containsKey("t_address"));
        assertThat(actual.get("t_order").get(0).getDataSourceName(), is("ds_0"));
        assertThat(actual.get("t_order_item").get(0).getDataSourceName(), is("single"));
        assertThat(actual.get("t_address").get(0).getDataSourceName(), is("broadcast"));
    }