[feat-#1455] fix hbase bug & add hbase ttl (#1456)

* [feat-##1455] fix hbase bug & add hbase ttl

* [feat-##1455] update ttl value set

Author: liumengkai

chunjun-connectors/chunjun-connector-hbase-base/src/main/java/com/dtstack/chunjun/connector/hbase/conf/HBaseConf.java

@@ -61,6 +61,7 @@ public class HBaseConf extends ChunJunCommonConf {
     private String rowkeyExpress;
     private Integer versionColumnIndex;
     private String versionColumnValue;
+    private Long ttl;
 
     public String getEncoding() {
         return encoding;
@@ -181,4 +182,12 @@ public List<FieldConf> getColumnMetaInfos() {
     public void setColumnMetaInfos(List<FieldConf> columnMetaInfos) {
         this.columnMetaInfos = columnMetaInfos;
     }
+
+    public void setTTL(Long ttl) {
+        this.ttl = ttl;
+    }
+
+    public Long getTtl() {
+        return ttl;
+    }
 }

chunjun-connectors/chunjun-connector-hbase-base/src/main/java/com/dtstack/chunjun/connector/hbase/converter/HBaseColumnConverter.java

@@ -66,6 +66,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static com.dtstack.chunjun.connector.hbase.HBaseTypeUtils.MAX_TIMESTAMP_PRECISION;
 import static com.dtstack.chunjun.connector.hbase.HBaseTypeUtils.MIN_TIMESTAMP_PRECISION;
@@ -211,6 +212,9 @@ public Mutation toExternal(RowData rowData, Mutation output) throws Exception {
             put = new Put(rowkey, version);
         }
 
+        put.setTTL(
+                Optional.ofNullable(hBaseConf.getTtl()).orElseGet(() -> (long) Integer.MAX_VALUE));
+
         for (int i = 0; i < rowData.getArity(); i++) {
             if (rowKeyIndex == i || columnConfigIndex.contains(i)) {
                 continue;
@@ -224,13 +228,13 @@ public Mutation toExternal(RowData rowData, Mutation output) throws Exception {
             toExternalConverters.get(i).serialize(rowData, i, put);
             if (i == rowData.getArity() - 1) {
                 for (int x = 0; x < familyAndQualifierBack.length; x++) {
-                    familyAndQualifier[x] =
-                            Arrays.copyOf(
-                                    familyAndQualifierBack[x], familyAndQualifierBack[x].length);
-                    if (x + 1 < familyAndQualifierBack.length
-                            && familyAndQualifierBack[x + 1] == null) {
-                        familyAndQualifier[x + 1] = null;
-                        x = x + 1;
+                    if (familyAndQualifierBack[x] == null) {
+                        familyAndQualifier[x] = null;
+                    } else {
+                        familyAndQualifier[x] =
+                                Arrays.copyOf(
+                                        familyAndQualifierBack[x],
+                                        familyAndQualifierBack[x].length);
                     }
                 }
             }