Merge branch 'hotfix_1.10_4.0.x_33089' into 1.10_release_4.0.x

Author: gituser

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

@@ -26,6 +26,8 @@
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
 import java.util.List;
@@ -44,6 +46,8 @@
 
 public abstract class AbstractTableParser {
 
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractTableParser.class);
+
     private static final String PRIMARY_KEY = "primaryKey";
     private static final String NEST_JSON_FIELD_KEY = "nestFieldKey";
     private static final String CHAR_TYPE_NO_LENGTH = "CHAR";
@@ -115,8 +119,14 @@ public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo) {
          */
         if (tableInfo instanceof AbstractSideTableInfo) {
             tableInfo.getPrimaryKeys().stream()
-                    .filter(pk -> (!tableInfo.getFieldList().contains(pk) && pk.equals("rowkey")))
-                    .forEach(pk -> handleKeyNotHaveAlias(String.format("%s varchar", pk), tableInfo));
+                    .filter(pk -> (!tableInfo.getFieldList().contains(pk)))
+                    .forEach(pk -> {
+                        try {
+                            handleKeyNotHaveAlias(String.format("%s varchar", pk.trim()), tableInfo);
+                        } catch (Exception e) {
+                            LOG.error(String.format("Add primary key to field list failed. Reason: %s", e.getMessage()));
+                        }
+                    });
         }
 
         tableInfo.finish();

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/AbstractRowKeyModeDealer.java

@@ -55,8 +55,6 @@ public abstract class AbstractRowKeyModeDealer {
 
     protected List<FieldInfo> outFieldInfoList;
 
-    protected static final String ROWKEY = "rowkey";
-
     //key:Returns the value of the position, returns the index values ​​in the input data
     protected Map<Integer, Integer> inFieldIndex = Maps.newHashMap();
 

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/PreRowKeyModeDealerDealer.java

@@ -111,7 +111,7 @@ private String dealOneRow(ArrayList<ArrayList<KeyValue>> args, String rowKeyStr,
                         List<Object> sideVal = Lists.newArrayList();
                         for (String key : colNames) {
 
-                            if (ROWKEY.equalsIgnoreCase(key)) {
+                            if (!sideMap.containsKey(key)) {
                                 sideVal.add(rowKeyStr);
                                 continue;
                             }

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/RowKeyEqualModeDealer.java

@@ -85,7 +85,7 @@ public void asyncGetData(String tableName, String rowKeyStr, BaseRow input, Resu
                         List<Object> sideVal = Lists.newArrayList();
                         for(String key : colNames){
 
-                            if (ROWKEY.equalsIgnoreCase(key)) {
+                            if (!sideMap.containsKey(key)) {
                                 sideVal.add(rowKeyStr);
                                 continue;
                             }