(shardingsphere) branch master updated: CDC support broadcast table (#28999)

[zhonghongsheng]

This is an automated email from the ASF dual-hosted git repository.

zhonghongsheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git


The following commit(s) were added to refs/heads/master by this push:
     new e66222ad6c1 CDC support broadcast table (#28999)
e66222ad6c1 is described below

commit e66222ad6c1ca3b97f3f2adea7db76b1562d0302
Author: Xinze Guo <101622833+aze...@users.noreply.github.com>
AuthorDate: Fri Nov 10 17:23:56 2023 +0800

    CDC support broadcast table (#28999)
---
 kernel/data-pipeline/scenario/cdc/core/pom.xml                     | 5 +++++
 .../shardingsphere/data/pipeline/cdc/util/CDCDataNodeUtils.java    | 7 +++++++
 .../shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java  | 5 +++++
 3 files changed, 17 insertions(+)

diff --git a/kernel/data-pipeline/scenario/cdc/core/pom.xml 
b/kernel/data-pipeline/scenario/cdc/core/pom.xml
index 0baab795693..57140ae56d2 100644
--- a/kernel/data-pipeline/scenario/cdc/core/pom.xml
+++ b/kernel/data-pipeline/scenario/cdc/core/pom.xml
@@ -37,6 +37,11 @@
             <artifactId>shardingsphere-data-pipeline-cdc-protocol</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.shardingsphere</groupId>
+            <artifactId>shardingsphere-broadcast-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         
         <dependency>
             <groupId>org.apache.shardingsphere</groupId>
diff --git 
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCDataNodeUtils.java
 
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCDataNodeUtils.java
index ebd2fdcb73f..7231d64ecce 100644
--- 
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCDataNodeUtils.java
+++ 
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCDataNodeUtils.java
@@ -19,6 +19,7 @@ package org.apache.shardingsphere.data.pipeline.cdc.util;
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
+import org.apache.shardingsphere.broadcast.rule.BroadcastRule;
 import 
org.apache.shardingsphere.data.pipeline.core.exception.param.PipelineInvalidParameterException;
 import org.apache.shardingsphere.infra.datanode.DataNode;
 import 
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
@@ -28,6 +29,7 @@ import org.apache.shardingsphere.single.rule.SingleRule;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -51,6 +53,7 @@ public final class CDCDataNodeUtils {
         Optional<ShardingRule> shardingRule = 
database.getRuleMetaData().findSingleRule(ShardingRule.class);
         Optional<SingleRule> singleRule = 
database.getRuleMetaData().findSingleRule(SingleRule.class);
         Map<String, List<DataNode>> result = new HashMap<>();
+        Optional<BroadcastRule> broadcastRule = 
database.getRuleMetaData().findSingleRule(BroadcastRule.class);
         // TODO support virtual data source name
         for (String each : tableNames) {
             if (singleRule.isPresent() && 
singleRule.get().getAllDataNodes().containsKey(each)) {
@@ -62,6 +65,10 @@ public final class CDCDataNodeUtils {
                 result.put(each, tableRule.getActualDataNodes());
                 continue;
             }
+            if (broadcastRule.isPresent() && 
broadcastRule.get().findFirstActualTable(each).isPresent()) {
+                result.put(each, 
Collections.singletonList(broadcastRule.get().getTableDataNodes().get(each).iterator().next()));
+                continue;
+            }
             throw new PipelineInvalidParameterException(String.format("Not 
find actual data nodes of `%s`", each));
         }
         return result;
diff --git 
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java
 
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java
index 5d0d7dea417..34f49f7ddb5 100644
--- 
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java
+++ 
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java
@@ -103,6 +103,7 @@ class CDCE2EIT {
                 containerComposer.registerStorageUnit(each);
             }
             createOrderTableRule(containerComposer);
+            createBroadcastRule(containerComposer);
             try (Connection connection = 
containerComposer.getProxyDataSource().getConnection()) {
                 initSchemaAndTable(containerComposer, connection, 3);
             }
@@ -153,6 +154,10 @@ class CDCE2EIT {
         Awaitility.await().atMost(20L, TimeUnit.SECONDS).pollInterval(2L, 
TimeUnit.SECONDS).until(() -> !containerComposer.queryForListWithLog("SHOW 
SHARDING TABLE RULE t_order").isEmpty());
     }
     
+    private void createBroadcastRule(final PipelineContainerComposer 
containerComposer) throws SQLException {
+        containerComposer.proxyExecuteWithLog("CREATE BROADCAST TABLE RULE 
t_address", 2);
+    }
+    
     private void initSchemaAndTable(final PipelineContainerComposer 
containerComposer, final Connection connection, final int sleepSeconds) throws 
SQLException {
         containerComposer.createSchema(connection, sleepSeconds);
         String sql = 
containerComposer.getExtraSQLCommand().getCreateTableOrder(SOURCE_TABLE_NAME);