[Feature][Connector-V2][Hbase] Introduce hbase sink connector

config/plugin_config

@@ -70,4 +70,6 @@ connector-slack
 connector-socket
 connector-starrocks
 connector-tablestore
+connector-selectdb-cloud
+connector-hbase
 --end--

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

@@ -0,0 +1,121 @@
+# Hbase
+
+> Hbase sink connector
+
+## Description
+
+Output data to Hbase
+
+## Key features
+
+- [ ] [exactly-once](../../concept/connector-v2-features.md)
+
+## Options
+
+| name               | type    | required | default value   |
+|--------------------|---------|----------|-----------------|
+| zookeeper_quorum   | string  | yes      | -               |
+| table              | string  | yes      | -               |
+| rowkey_column      | list    | yes      | -               |
+| family_name        | config  | yes      | -               |
+| rowkey_delimiter   | string  | no       | ""              |
+| version_column     | string  | no       | -               |
+| null_mode          | string  | no       | skip            |
+| wal_write          | boolean | yes      | false           |
+| write_buffer_size  | string  | no       | 8 * 1024 * 1024 |
+| encoding           | string  | no       | utf8            |
+| hbase_extra_config | string  | no       | -               |
+| common-options     |         | no       | -               |
+
+### zookeeper_quorum [string]
+
+The zookeeper cluster host of hbase, example: "hadoop001:2181,hadoop002:2181,hadoop003:2181"
+
+### table [string]
+
+The table name you want to write, example: "seatunnel"
+
+### rowkey_column [list]
+
+The column name list of row keys, example: ["id", "uuid"]
+
+### family_name [config]
+
+The family name mapping of fields. For example the row from upstream like the following shown:
+
+| id  | name          | age |
+|-----|---------------|-----|
+| 1   | tyrantlucifer | 27  |
+
+id as the row key and other fields written to the different families, you can assign
+
+family_name {
+    name = "info1"
+    age = "info2"
+}
+
+this means that `name` will be written to the family `info1` and the `age` will be written to the family `info2`
+
+if you want other fields written to the same family, you can assign
+
+family_name {
+    all_columns = "info"
+}
+
+this means that all fields will be written to the family `info`
+
+### rowkey_delimiter [string]
+
+The delimiter of joining multi row keys, default `""`
+
+### version_column [string]
+
+The version column name, you can use it to assign timestamp for hbase record
+
+### null_mode [double]
+
+The mode of writing null value, support [`skip`, `empty`], default `skip`
+
+- skip: When the field is null, connector will not write this field to hbase
+- empty: When the field is null, connector will write generate empty value for this field
+
+### wal_write [boolean]
+
+The wal log write flag, default `false`
+
+### write_buffer_size [int]
+
+The write buffer size of hbase client, default `8 * 1024 * 1024`
+
+### encoding [string]
+
+The encoding of string field, support [`utf8`, `gbk`], default `utf8`
+
+### hbase_extra_config [config]
+
+The extra configuration of hbase
+
+### common options
+
+Sink plugin common parameters, please refer to [Sink Common Options](common-options.md) for details
+
+## Example
+
+```hocon
+
+Hbase {
+  zookeeper_quorum = "hadoop001:2181,hadoop002:2181,hadoop003:2181"
+  table = "seatunnel_test"
+  rowkey_column = ["name"]
+  family_name {
+    all_columns = seatunnel
+  }
+}
+
+```
+
+## Changelog
+
+### next version
+
+- Add hbase sink connector ([4049](https://github.com/apache/incubator-seatunnel/pull/4049))

plugin-mapping.properties

@@ -170,4 +170,5 @@ seatunnel.source.TDengine = connector-tdengine
 seatunnel.sink.TDengine = connector-tdengine
 seatunnel.source.Persistiq = connector-http-persistiq
 seatunnel.sink.SelectDBCloud = connector-selectdb-cloud
+seatunnel.sink.Hbase = connector-hbase
 

release-note.md

@@ -12,6 +12,7 @@
 - [ALL]Add FieldMapper Transform #3781
 ### Connectors
 - [Elasticsearch] Support https protocol & compatible with opensearch
+- [Hbase] Add hbase sink connector #4049
 ### Formats
 - [Canal]Support read canal format message #3950
 

seatunnel-common/src/main/java/org/apache/seatunnel/common/config/TypesafeConfigUtils.java

@@ -23,6 +23,7 @@
 
 import lombok.NonNull;
 
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -108,4 +109,12 @@ public static List<? extends Config> getConfigList(Config config,
                                                        @NonNull List<? extends Config> defaultValue) {
         return config.hasPath(configKey) ? config.getConfigList(configKey) : defaultValue;
     }
+
+    public static Map<String, String> configToMap(Config config) {
+        Map<String, String> configMap = new HashMap<>();
+        config.entrySet().forEach(entry -> {
+            configMap.put(entry.getKey(), entry.getValue().unwrapped().toString());
+        });
+        return configMap;
+    }
 }

seatunnel-connectors-v2/connector-hbase/pom.xml

@@ -0,0 +1,52 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You under the Apache License, Version 2.0
+    (the "License"); you may not use this file except in compliance with
+    the License.  You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.seatunnel</groupId>
+        <artifactId>seatunnel-connectors-v2</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>connector-hbase</artifactId>
+
+    <properties>
+        <hbase.version>2.4.10</hbase.version>
+    </properties>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hbase</groupId>
+            <artifactId>hbase-client</artifactId>
+            <version>${hbase.version}</version>
+        </dependency>
+
+    </dependencies>
+
+</project>

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

@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.hbase.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+import java.util.List;
+import java.util.Map;
+
+public class HbaseConfig {
+
+    private static final Integer DEFAULT_BUFFER_SIZE = 8 * 1024 * 1024;
+
+    public static final Option<String> ZOOKEEPER_QUORUM = Options.key("zookeeper_quorum")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Hbase zookeeper quorum");
+
+    public static final Option<String> TABLE = Options.key("table")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Hbase table name");
+
+    public static final Option<List<String>> ROWKEY_COLUMNS = Options.key("rowkey_column")
+            .listType()
+            .noDefaultValue()
+            .withDescription("Hbase rowkey column");
+
+    public static final Option<String> ROWKEY_DELIMITER = Options.key("rowkey_delimiter")
+            .stringType()
+            .defaultValue("")
+            .withDescription("Hbase rowkey join delimiter");
+
+    public static final Option<String> VERSION_COLUMN = Options.key("version_column")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Hbase record version column used for assigning timestamp of records");
+
+    public static final Option<NullMode> NULL_MODE = Options.key("null_mode")
+            .enumType(NullMode.class)
+            .defaultValue(NullMode.SKIP)
+            .withDescription("The processing mode for writing null values");
+
+    public static final Option<Boolean> WAL_WRITE = Options.key("wal_write")
+            .booleanType()
+            .defaultValue(false)
+            .withDescription("The flag of whether write wal log");
+
+    public static final Option<Integer> WRITE_BUFFER_SIZE = Options.key("write_buffer_size")
+            .intType()
+            .defaultValue(DEFAULT_BUFFER_SIZE)
+            .withDescription("Hbase client write buffer size");
+
+    public static final Option<EnCoding> ENCODING = Options.key("encoding")
+            .enumType(EnCoding.class)
+            .defaultValue(EnCoding.UTF8)
+            .withDescription("Hbase record encoding");
+
+    public static final Option<Map<String, String>> FAMILY_NAME = Options.key("family_name")
+            .mapType()
+            .noDefaultValue()
+            .withDescription("Hbase column family name");
+
+    public static final Option<Map<String, String>> HBASE_EXTRA_CONFIG = Options.key("hbase_extra_config")
+            .mapType()
+            .noDefaultValue()
+            .withDescription("Hbase extra config");
+
+    public enum NullMode {
+        SKIP,
+        EMPTY;
+    }
+
+    public enum EnCoding {
+        UTF8,
+        GBK;
+    }
+
+    private HbaseConfig() {}
+}