[feature][connector-v2-hbase-sink] Support Connector v2 HBase sink TT…

…L data writing (apache#7116)
chaorongzhi · Aug 21, 2024 · 67c1d81 · 67c1d81
1 parent 42b9d77
commit 67c1d81
Show file tree

Hide file tree

Showing 5 changed files with 26 additions and 0 deletions.
diff --git a/docs/en/connector-v2/sink/Hbase.md b/docs/en/connector-v2/sink/Hbase.md
@@ -26,6 +26,7 @@ Output data to Hbase
 | encoding           | string  | no       | utf8            |
 | hbase_extra_config | string  | no       | -               |
 | common-options     |         | no       | -               |
+| ttl                | long    | no       | -               |
 
 ### zookeeper_quorum [string]
 
@@ -95,6 +96,10 @@ The encoding of string field, support [`utf8`, `gbk`], default `utf8`
 
 The extra configuration of hbase
 
+### ttl [long]
+
+Hbase writes data TTL time, the default is based on the TTL set in the table, unit: milliseconds
+
 ### common options
 
 Sink plugin common parameters, please refer to [Sink Common Options](common-options.md) for details

diff --git a/docs/zh/connector-v2/sink/Hbase.md b/docs/zh/connector-v2/sink/Hbase.md
@@ -26,6 +26,7 @@
 | encoding           | string  | no   | utf8            |
 | hbase_extra_config | string  | no   | -               |
 | common-options     |         | no   | -               |
+| ttl                | long    | no   | -               |
 
 ### zookeeper_quorum [string]
 
@@ -95,6 +96,10 @@ hbase 客户端的写入缓冲区大小，默认 8 * 1024 * 1024
 
 hbase扩展配置
 
+### ttl [long]
+
+hbase 写入数据 TTL 时间，默认以表设置的TTL为准，单位毫秒
+
 ### 常见选项
 
 Sink 插件常用参数，详见 Sink 常用选项 [Sink Common Options](common-options.md)

diff --git a/...ase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/config/HbaseConfig.java b/...ase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/config/HbaseConfig.java
@@ -97,6 +97,13 @@ public class HbaseConfig {
                     .noDefaultValue()
                     .withDescription("Hbase extra config");
 
+    public static final Option<Long> HBASE_TTL_CONFIG =
+            Options.key("ttl")
+                    .longType()
+                    .defaultValue(-1L)
+                    .withDescription(
+                            "The expiration time configuration for writing hbase data. The default value is -1, indicating no expiration time.");
+
     public enum NullMode {
         SKIP,
         EMPTY;

diff --git a/...src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/config/HbaseParameters.java b/...src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/config/HbaseParameters.java
@@ -31,6 +31,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.ENCODING;
 import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.FAMILY_NAME;
 import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.HBASE_EXTRA_CONFIG;
+import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.HBASE_TTL_CONFIG;
 import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.NULL_MODE;
 import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.QUERY_COLUMNS;
 import static org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseConfig.ROWKEY_COLUMNS;
@@ -59,6 +60,8 @@ public class HbaseParameters implements Serializable {
 
     private Map<String, String> hbaseExtraConfig;
 
+    @Builder.Default private Long ttl = HBASE_TTL_CONFIG.defaultValue();
+
     @Builder.Default private String rowkeyDelimiter = ROWKEY_DELIMITER.defaultValue();
 
     @Builder.Default private HbaseConfig.NullMode nullMode = NULL_MODE.defaultValue();
@@ -80,6 +83,9 @@ public static HbaseParameters buildWithConfig(Config pluginConfig) {
                 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()));
         }

diff --git a/...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/sink/HbaseSinkWriter.java b/...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/sink/HbaseSinkWriter.java
@@ -117,6 +117,9 @@ private Put convertRowToPut(SeaTunnelRow row) {
             timestamp = (Long) row.getField(versionColumnIndex);
         }
         Put put = new Put(rowkey, timestamp);
+        if (hbaseParameters.getTtl() != -1 && hbaseParameters.getTtl() > 0) {
+            put.setTTL(hbaseParameters.getTtl());
+        }
         if (!hbaseParameters.isWalWrite()) {
             put.setDurability(Durability.SKIP_WAL);
         }