Merge branch 'feat_1.8_asyncException_mergeDev' into feat_1.8_asyncException_mergeTest

Author: dapeng

core/src/main/java/com/dtstack/flink/sql/side/BaseAsyncReqRow.java

@@ -161,7 +161,8 @@ protected void preInvoke(CRow input, ResultFuture<CRow> resultFuture){
     }
 
     @Override
-    public void asyncInvoke(CRow input, ResultFuture<CRow> resultFuture) throws Exception {
+    public void asyncInvoke(CRow row, ResultFuture<CRow> resultFuture) throws Exception {
+        CRow input = new CRow(Row.copy(row.row()), row.change());
         preInvoke(input, resultFuture);
         Map<String, Object> inputParams = parseInputParam(input);
         if(MapUtils.isEmpty(inputParams)){

redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java

@@ -144,7 +144,7 @@ public void accept(Map<String, String> values) {
                         dealCacheData(key,CacheObj.buildCacheObj(ECacheContentType.MultiLine, values));
                         resultFuture.complete(Collections.singleton(new CRow(row, input.change())));
                     } catch (Exception e) {
-                        dealFillDataError(resultFuture, e, input);
+                        dealFillDataError(input, resultFuture, e);
                     }
                 } else {
                     dealMissKey(input, resultFuture);