【fix] use LinkedHashMap replace Map, LinkedList replace set

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfo.java

@@ -49,7 +49,7 @@ public abstract class AbstractTableInfo implements Serializable {
     private final List<String> fieldList = Lists.newArrayList();
 
     /**key:别名, value: realField */
-    private Map<String, String> physicalFields = Maps.newHashMap();
+    private Map<String, String> physicalFields = Maps.newLinkedHashMap();
 
     private final List<String> fieldTypeList = Lists.newArrayList();
 

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java

@@ -28,7 +28,11 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Preconditions;
-import org.apache.hadoop.hbase.*;
+import org.apache.hadoop.hbase.AuthUtil;
+import org.apache.hadoop.hbase.ChoreService;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.ScheduledChore;
+import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
 import org.apache.hadoop.hbase.client.Delete;
@@ -42,9 +46,10 @@
 import java.io.File;
 import java.io.IOException;
 import java.security.PrivilegedAction;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 /**
  * @author: jingzhen@dtstack.com
@@ -61,7 +66,7 @@ public class HbaseOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
     private String[] columnNames;
     private String updateMode;
     private String[] columnTypes;
-    private Map<String, String> columnNameFamily;
+    private LinkedHashMap<String, String> columnNameFamily;
 
     private boolean kerberosAuthEnable;
     private String regionserverKeytabFile;
@@ -129,21 +134,18 @@ private void openKerberosConn() throws Exception {
 
         UserGroupInformation userGroupInformation = HbaseConfigUtils.loginAndReturnUGI(conf, clientPrincipal, clientKeytabFile);
         org.apache.hadoop.conf.Configuration finalConf = conf;
-        conn = userGroupInformation.doAs(new PrivilegedAction<Connection>() {
-            @Override
-            public Connection run() {
-                try {
-                    ScheduledChore authChore = AuthUtil.getAuthChore(finalConf);
-                    if (authChore != null) {
-                        choreService = new ChoreService("hbaseKerberosSink");
-                        choreService.scheduleChore(authChore);
-                    }
-
-                    return ConnectionFactory.createConnection(finalConf);
-                } catch (IOException e) {
-                    LOG.error("Get connection fail with config:{}", finalConf);
-                    throw new RuntimeException(e);
+        conn = userGroupInformation.doAs((PrivilegedAction<Connection>) () -> {
+            try {
+                ScheduledChore authChore = AuthUtil.getAuthChore(finalConf);
+                if (authChore != null) {
+                    choreService = new ChoreService("hbaseKerberosSink");
+                    choreService.scheduleChore(authChore);
                 }
+
+                return ConnectionFactory.createConnection(finalConf);
+            } catch (IOException e) {
+                LOG.error("Get connection fail with config:{}", finalConf);
+                throw new RuntimeException(e);
             }
         });
     }
@@ -304,7 +306,7 @@ public HbaseOutputFormatBuilder setColumnTypes(String[] columnTypes) {
             return this;
         }
 
-        public HbaseOutputFormatBuilder setColumnNameFamily(Map<String, String> columnNameFamily) {
+        public HbaseOutputFormatBuilder setColumnNameFamily(LinkedHashMap<String, String> columnNameFamily) {
             format.columnNameFamily = columnNameFamily;
             return this;
         }
@@ -355,8 +357,8 @@ public HbaseOutputFormat finish() {
             String[] qualifiers = new String[format.columnNames.length];
 
             if (format.columnNameFamily != null) {
-                Set<String> keySet = format.columnNameFamily.keySet();
-                String[] columns = keySet.toArray(new String[keySet.size()]);
+                List<String> keyList = new LinkedList<>(format.columnNameFamily.keySet());
+                String[] columns = keyList.toArray(new String[0]);
                 for (int i = 0; i < columns.length; ++i) {
                     String col = columns[i];
                     String[] part = col.split(":");

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseSink.java

@@ -35,7 +35,7 @@
 import org.apache.flink.table.sinks.TableSink;
 import org.apache.flink.types.Row;
 
-import java.util.Map;
+import java.util.LinkedHashMap;
 
 /**
  * Date: 2018/09/14
@@ -45,7 +45,7 @@
 public class HbaseSink implements RetractStreamTableSink<Row>, IStreamSinkGener<HbaseSink> {
 
     protected String[] fieldNames;
-    protected Map<String, String> columnNameFamily;
+    protected LinkedHashMap<String, String> columnNameFamily;
     TypeInformation<?>[] fieldTypes;
     protected String zookeeperQuorum;
     protected String port;

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/table/HbaseSinkParser.java

@@ -97,7 +97,6 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
 
     public void parseFieldsInfo(String fieldsInfo, HbaseTableInfo tableInfo){
         List<String> fieldRows = DtStringUtil.splitIgnoreQuota(fieldsInfo, ',');
-        Map<String, String> columnFamilies = new LinkedHashMap<>();
         for(String fieldRow : fieldRows){
             fieldRow = fieldRow.trim();
 
@@ -128,8 +127,8 @@ public void parseFieldsInfo(String fieldsInfo, HbaseTableInfo tableInfo){
         tableInfo.finish();
     }
 
-    private Map<String, String> parseColumnFamily(Map<String, String> physicalFieldMap){
-        Map<String, String> columnFamiles = Maps.newLinkedHashMap();
+    private LinkedHashMap<String, String> parseColumnFamily(Map<String, String> physicalFieldMap){
+        LinkedHashMap<String, String> columnFamiles = Maps.newLinkedHashMap();
         physicalFieldMap.values().forEach(x -> {
             String[] columnFamily = StringUtils.split(x.trim(), ":");
             columnFamiles.put(x, columnFamily[1]);

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/table/HbaseTableInfo.java

@@ -25,6 +25,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Maps;
 
+import java.util.LinkedHashMap;
 import java.util.Map;
 
 /**
@@ -44,7 +45,7 @@ public class HbaseTableInfo extends AbstractTargetTableInfo {
 
     private String rowkey;
 
-    private Map<String, String> columnNameFamily;
+    private LinkedHashMap<String, String> columnNameFamily;
 
     private String[] columnNames;
 
@@ -116,11 +117,11 @@ public void setRowkey(String rowkey) {
         this.rowkey = rowkey;
     }
 
-    public Map<String, String> getColumnNameFamily() {
+    public LinkedHashMap<String, String> getColumnNameFamily() {
         return columnNameFamily;
     }
 
-    public void setColumnNameFamily(Map<String, String> columnNameFamily) {
+    public void setColumnNameFamily(LinkedHashMap<String, String> columnNameFamily) {
         this.columnNameFamily = columnNameFamily;
     }