Merge branch '1.10_release_4.0.x' into 1.10_release_4.1.x

Author: gituser

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
@@ -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);
             }
         });
     }
@@ -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/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();
 
@@ -129,7 +128,7 @@ public void parseFieldsInfo(String fieldsInfo, HbaseTableInfo tableInfo){
     }
 
     private Map<String, String> parseColumnFamily(Map<String, String> physicalFieldMap){
-        Map<String, String> columnFamiles = Maps.newHashMap();
+        Map<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;
 
 /**