[hotfix-35067][rdb] share rdb async connect rollback

Author: simenliuxing

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -84,7 +84,7 @@ public class RdbAsyncReqRow extends BaseAsyncReqRow {
 
     public final static String PREFERRED_TEST_QUERY_SQL = "SELECT 1 FROM DUAL";
 
-    private static SQLClient rdbSqlClient;
+    private transient SQLClient rdbSqlClient;
 
     private AtomicBoolean connectionStatus = new AtomicBoolean(true);
 
@@ -126,7 +126,7 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, BaseRow input, Re
             Thread.sleep(100);
         }
         Map<String, Object> params = formatInputParam(inputParams);
-        executor.execute(() -> connectWithRetry(params, input, resultFuture, RdbAsyncReqRow.rdbSqlClient));
+        executor.execute(() -> connectWithRetry(params, input, resultFuture, rdbSqlClient));
     }
 
     protected void asyncQueryData(Map<String, Object> inputParams,
@@ -280,8 +280,8 @@ public BaseRow fillData(BaseRow input, Object line) {
     @Override
     public void close() throws Exception {
         super.close();
-        if (RdbAsyncReqRow.rdbSqlClient != null) {
-            RdbAsyncReqRow.rdbSqlClient.close();
+        if (rdbSqlClient != null) {
+            rdbSqlClient.close();
         }
 
         if (executor != null) {
@@ -291,7 +291,7 @@ public void close() throws Exception {
     }
 
     public void setRdbSqlClient(SQLClient rdbSqlClient) {
-        RdbAsyncReqRow.rdbSqlClient = rdbSqlClient;
+        this.rdbSqlClient = rdbSqlClient;
     }
 
     private void handleQuery(SQLConnection connection, Map<String, Object> inputParams, BaseRow input, ResultFuture<BaseRow> resultFuture) {