Merge remote-tracking branch 'origin/hotfix_1.8_3.10.x_25279' into 1.8_zy_3.10.x

Author: dapeng

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

@@ -89,6 +89,7 @@ public HbaseAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldIn
 
     @Override
     public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
         SideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
         HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;
         ExecutorService executorService =new ThreadPoolExecutor(DEFAULT_POOL_SIZE, DEFAULT_POOL_SIZE,
@@ -147,7 +148,7 @@ public void asyncInvoke(CRow input, ResultFuture<CRow> resultFuture) throws Exce
                     return;
                 } else if (ECacheContentType.SingleLine == val.getType()) {
                     try {
-                        Row row = fillData(inputCopy.row(), val);
+                        Row row = fillData(inputCopy.row(), val.getContent());
                         resultFuture.complete(Collections.singleton(new CRow(row, inputCopy.change())));
                     } catch (Exception e) {
                         dealFillDataError(resultFuture, e, inputCopy);

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

@@ -95,7 +95,7 @@ public void asyncGetData(String tableName, String rowKeyStr, CRow input, ResultF
 
                         Row row = fillData(input.row(), sideVal);
                         if(openCache){
-                            sideCache.putCache(rowKeyStr, CacheObj.buildCacheObj(ECacheContentType.SingleLine, row));
+                            sideCache.putCache(rowKeyStr, CacheObj.buildCacheObj(ECacheContentType.SingleLine, sideVal));
                         }
                         resultFuture.complete(Collections.singleton(new CRow(row, input.change())));
                     } catch (Exception e) {

hbase/hbase-side/hbase-side-core/src/main/java/com/dtstack/flink/sql/side/hbase/table/HbaseSideParser.java

@@ -61,6 +61,7 @@ public HbaseSideParser() {
     public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
         HbaseSideTableInfo hbaseTableInfo = new HbaseSideTableInfo();
         hbaseTableInfo.setName(tableName);
+        parseCacheProp(hbaseTableInfo, props);
         parseFieldsInfo(fieldsInfo, hbaseTableInfo);
         hbaseTableInfo.setTableName((String) props.get(TABLE_NAME_KEY.toLowerCase()));
         hbaseTableInfo.setParallelism(MathUtil.getIntegerVal(props.get(PARALLELISM_KEY.toLowerCase())));