async row copy

Author: todd5167

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java

@@ -161,7 +161,7 @@ private void connCassandraDB(CassandraSideTableInfo tableInfo) {
 
     @Override
     public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {
-        Row inputRow = input;
+        Row inputRow = Row.copy(input);
         JsonArray inputParams = new JsonArray();
         StringBuffer stringBuffer = new StringBuffer();
         String sqlWhere = " where ";

kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncReqRow.java

@@ -121,7 +121,7 @@ private void connKuDu() throws KuduException {
 
     @Override
     public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {
-        Row inputRow = input;
+        Row inputRow = Row.copy(input);
         //scannerBuilder 设置为null重新加载过滤条件
         scannerBuilder = null;
         connKuDu();

mongo/mongo-side/mongo-async-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAsyncReqRow.java

@@ -124,7 +124,7 @@ public void connMongoDB() throws Exception {
 
     @Override
     public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {
-        Row inputRow = input;
+        Row inputRow = Row.copy(input);
         BasicDBObject basicDBObject = new BasicDBObject();
         for (int i = 0; i < sideInfo.getEqualFieldList().size(); i++) {
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);

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

@@ -117,7 +117,7 @@ public Row fillData(Row input, Object sideInput) {
 
     @Override
     public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {
-        Row inputRow = input;
+        Row inputRow = Row.copy(input);
         List<String> keyData = Lists.newLinkedList();
         for (int i = 0; i < sideInfo.getEqualValIndex().size(); i++) {
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);