[Feature][Connector-V2] Support multi-table sink feature for HBase

docs/en/connector-v2/sink/Hbase.md

@@ -116,6 +116,79 @@ Hbase {
     all_columns = seatunnel
   }
 }
+
+```
+
+### Multiple Table
+
+```hocon
+env {
+  # You can set engine configuration here
+  execution.parallelism = 1
+  job.mode = "BATCH"
+}
+
+source {
+  FakeSource {
+    tables_configs = [
+       {
+        schema = {
+          table = "hbase_sink_1"
+         fields {
+                    name = STRING
+                    c_string = STRING
+                    c_double = DOUBLE
+                    c_bigint = BIGINT
+                    c_float = FLOAT
+                    c_int = INT
+                    c_smallint = SMALLINT
+                    c_boolean = BOOLEAN
+                    time = BIGINT
+           }
+        }
+            rows = [
+              {
+                kind = INSERT
+                fields = ["label_1", "sink_1", 4.3, 200, 2.5, 2, 5, true, 1627529632356]
+              }
+              ]
+       },
+       {
+       schema = {
+         table = "hbase_sink_2"
+              fields {
+                    name = STRING
+                    c_string = STRING
+                    c_double = DOUBLE
+                    c_bigint = BIGINT
+                    c_float = FLOAT
+                    c_int = INT
+                    c_smallint = SMALLINT
+                    c_boolean = BOOLEAN
+                    time = BIGINT
+              }
+       }
+           rows = [
+             {
+               kind = INSERT
+               fields = ["label_2", "sink_2", 4.3, 200, 2.5, 2, 5, true, 1627529632357]
+             }
+             ]
+      }
+    ]
+  }
+}
+
+sink {
+  Hbase {
+    zookeeper_quorum = "hadoop001:2181,hadoop002:2181,hadoop003:2181"
+    table = "${table_name}"
+    rowkey_column = ["name"]
+    family_name {
+      all_columns = info
+    }
+  }
+}
 ```
 
 ## Writes To The Specified Column Family

docs/zh/connector-v2/sink/Hbase.md

@@ -119,6 +119,78 @@ Hbase {
 
 ```
 
+### 写入多表
+
+```hocon
+env {
+  # You can set engine configuration here
+  execution.parallelism = 1
+  job.mode = "BATCH"
+}
+
+source {
+  FakeSource {
+    tables_configs = [
+       {
+        schema = {
+          table = "hbase_sink_1"
+         fields {
+                    name = STRING
+                    c_string = STRING
+                    c_double = DOUBLE
+                    c_bigint = BIGINT
+                    c_float = FLOAT
+                    c_int = INT
+                    c_smallint = SMALLINT
+                    c_boolean = BOOLEAN
+                    time = BIGINT
+           }
+        }
+            rows = [
+              {
+                kind = INSERT
+                fields = ["label_1", "sink_1", 4.3, 200, 2.5, 2, 5, true, 1627529632356]
+              }
+              ]
+       },
+       {
+       schema = {
+         table = "hbase_sink_2"
+              fields {
+                    name = STRING
+                    c_string = STRING
+                    c_double = DOUBLE
+                    c_bigint = BIGINT
+                    c_float = FLOAT
+                    c_int = INT
+                    c_smallint = SMALLINT
+                    c_boolean = BOOLEAN
+                    time = BIGINT
+              }
+       }
+           rows = [
+             {
+               kind = INSERT
+               fields = ["label_2", "sink_2", 4.3, 200, 2.5, 2, 5, true, 1627529632357]
+             }
+             ]
+      }
+    ]
+  }
+}
+
+sink {
+  Hbase {
+    zookeeper_quorum = "hadoop001:2181,hadoop002:2181,hadoop003:2181"
+    table = "${table_name}"
+    rowkey_column = ["name"]
+    family_name {
+      all_columns = info
+    }
+  }
+}
+```
+
 ## 写入指定列族
 
 ```hocon

seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/config/HbaseParameters.java

@@ -19,6 +19,7 @@
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.common.config.TypesafeConfigUtils;
 
 import lombok.Builder;
@@ -80,44 +81,25 @@ public class HbaseParameters implements Serializable {
 
     @Builder.Default private HbaseConfig.EnCoding enCoding = ENCODING.defaultValue();
 
-    public static HbaseParameters buildWithSinkConfig(Config pluginConfig) {
+    public static HbaseParameters buildWithConfig(ReadonlyConfig config) {
         HbaseParametersBuilder builder = HbaseParameters.builder();
 
         // required parameters
-        builder.zookeeperQuorum(pluginConfig.getString(ZOOKEEPER_QUORUM.key()));
-        builder.table(pluginConfig.getString(TABLE.key()));
-        builder.rowkeyColumns(pluginConfig.getStringList(ROWKEY_COLUMNS.key()));
-        builder.familyNames(
-                TypesafeConfigUtils.configToMap(pluginConfig.getConfig(FAMILY_NAME.key())));
-
-        // optional parameters
-        if (pluginConfig.hasPath(HBASE_TTL_CONFIG.key())) {
-            builder.ttl(pluginConfig.getLong(HBASE_TTL_CONFIG.key()));
-        }
-        if (pluginConfig.hasPath(ROWKEY_DELIMITER.key())) {
-            builder.rowkeyDelimiter(pluginConfig.getString(ROWKEY_DELIMITER.key()));
-        }
-        if (pluginConfig.hasPath(VERSION_COLUMN.key())) {
-            builder.versionColumn(pluginConfig.getString(VERSION_COLUMN.key()));
-        }
-        if (pluginConfig.hasPath(NULL_MODE.key())) {
-            String nullMode = pluginConfig.getString(NULL_MODE.key());
-            builder.nullMode(HbaseConfig.NullMode.valueOf(nullMode.toUpperCase()));
-        }
-        if (pluginConfig.hasPath(WAL_WRITE.key())) {
-            builder.walWrite(pluginConfig.getBoolean(WAL_WRITE.key()));
-        }
-        if (pluginConfig.hasPath(WRITE_BUFFER_SIZE.key())) {
-            builder.writeBufferSize(pluginConfig.getInt(WRITE_BUFFER_SIZE.key()));
-        }
-        if (pluginConfig.hasPath(ENCODING.key())) {
-            String encoding = pluginConfig.getString(ENCODING.key());
-            builder.enCoding(HbaseConfig.EnCoding.valueOf(encoding.toUpperCase()));
-        }
-        if (pluginConfig.hasPath(HBASE_EXTRA_CONFIG.key())) {
-            Config extraConfig = pluginConfig.getConfig(HBASE_EXTRA_CONFIG.key());
-            builder.hbaseExtraConfig(TypesafeConfigUtils.configToMap(extraConfig));
-        }
+        builder.zookeeperQuorum(config.get(ZOOKEEPER_QUORUM));
+        builder.rowkeyColumns(config.get(ROWKEY_COLUMNS));
+        builder.familyNames(config.get(FAMILY_NAME));
+
+        builder.table(config.get(TABLE));
+        builder.rowkeyDelimiter(config.get(ROWKEY_DELIMITER));
+        builder.versionColumn(config.get(VERSION_COLUMN));
+        String nullMode = String.valueOf(config.get(NULL_MODE));
+        builder.nullMode(HbaseConfig.NullMode.valueOf(nullMode.toUpperCase()));
+        builder.walWrite(config.get(WAL_WRITE));
+        builder.writeBufferSize(config.get(WRITE_BUFFER_SIZE));
+        String encoding = String.valueOf(config.get(ENCODING));
+        builder.enCoding(HbaseConfig.EnCoding.valueOf(encoding.toUpperCase()));
+        builder.hbaseExtraConfig(config.get(HBASE_EXTRA_CONFIG));
+        builder.ttl(config.get(HBASE_TTL_CONFIG));
         return builder.build();
     }
 

seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/sink/HbaseSink.java

@@ -19,33 +19,20 @@
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
-import org.apache.seatunnel.api.common.PrepareFailException;
-import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
-import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.sink.SupportMultiTableSink;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
-import org.apache.seatunnel.common.config.CheckConfigUtil;
-import org.apache.seatunnel.common.config.CheckResult;
-import org.apache.seatunnel.common.constants.PluginType;
 import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
-import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
 import org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseParameters;
-import org.apache.seatunnel.connectors.seatunnel.hbase.exception.HbaseConnectorException;
-
-import com.google.auto.service.AutoService;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
-import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.FAMILY_NAME;
-import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.ROWKEY_COLUMNS;
-import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.TABLE;
-import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.ZOOKEEPER_QUORUM;
-
-@AutoService(SeaTunnelSink.class)
-public class HbaseSink extends AbstractSimpleSink<SeaTunnelRow, Void> {
+public class HbaseSink extends AbstractSimpleSink<SeaTunnelRow, Void>
+        implements SupportMultiTableSink {
 
     private Config pluginConfig;
 
@@ -62,34 +49,9 @@ public String getPluginName() {
         return HbaseSinkFactory.IDENTIFIER;
     }
 
-    @Override
-    public void prepare(Config pluginConfig) throws PrepareFailException {
-        this.pluginConfig = pluginConfig;
-        CheckResult result =
-                CheckConfigUtil.checkAllExists(
-                        pluginConfig,
-                        ZOOKEEPER_QUORUM.key(),
-                        TABLE.key(),
-                        ROWKEY_COLUMNS.key(),
-                        FAMILY_NAME.key());
-        if (!result.isSuccess()) {
-            throw new HbaseConnectorException(
-                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
-                    String.format(
-                            "PluginName: %s, PluginType: %s, Message: %s",
-                            getPluginName(), PluginType.SINK, result.getMsg()));
-        }
-        this.hbaseParameters = HbaseParameters.buildWithSinkConfig(pluginConfig);
-        if (hbaseParameters.getFamilyNames().size() == 0) {
-            throw new HbaseConnectorException(
-                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
-                    "The corresponding field options should be configured and should not be empty Refer to the hbase sink document");
-        }
-    }
-
-    @Override
-    public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
-        this.seaTunnelRowType = seaTunnelRowType;
+    public HbaseSink(HbaseParameters hbaseParameters, CatalogTable catalogTable) {
+        this.hbaseParameters = hbaseParameters;
+        this.seaTunnelRowType = catalogTable.getTableSchema().toPhysicalRowDataType();
         for (String rowkeyColumn : hbaseParameters.getRowkeyColumns()) {
             this.rowkeyColumnIndexes.add(seaTunnelRowType.indexOf(rowkeyColumn));
         }
@@ -99,8 +61,7 @@ public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
     }
 
     @Override
-    public AbstractSinkWriter<SeaTunnelRow, Void> createWriter(SinkWriter.Context context)
-            throws IOException {
+    public HbaseSinkWriter createWriter(SinkWriter.Context context) throws IOException {
         return new HbaseSinkWriter(
                 seaTunnelRowType, hbaseParameters, rowkeyColumnIndexes, versionColumnIndex);
     }

seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/sink/HbaseSinkFactory.java

@@ -18,8 +18,13 @@
 package org.apache.seatunnel.connectors.seatunnel.hbase.sink;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.sink.SinkCommonOptions;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.connector.TableSink;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSinkFactory;
+import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
+import org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseParameters;
 
 import com.google.auto.service.AutoService;
 
@@ -50,6 +55,7 @@ public OptionRule optionRule() {
         return OptionRule.builder()
                 .required(ZOOKEEPER_QUORUM, TABLE, ROWKEY_COLUMNS, FAMILY_NAME)
                 .optional(
+                        SinkCommonOptions.MULTI_TABLE_SINK_REPLICA,
                         ROWKEY_DELIMITER,
                         VERSION_COLUMN,
                         NULL_MODE,
@@ -59,4 +65,11 @@ public OptionRule optionRule() {
                         HBASE_EXTRA_CONFIG)
                 .build();
     }
+
+    @Override
+    public TableSink createSink(TableSinkFactoryContext context) {
+        HbaseParameters hbaseParameters = HbaseParameters.buildWithConfig(context.getOptions());
+        CatalogTable catalogTable = context.getCatalogTable();
+        return () -> new HbaseSink(hbaseParameters, catalogTable);
+    }
 }