Pipe: Support syncing table model data between clusters (airgap) (#13828)

Author: SteveYurongSu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiver.java

@@ -27,6 +27,9 @@
 import org.apache.iotdb.commons.pipe.connector.payload.airgap.AirGapPseudoTPipeTransferRequest;
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
 import org.apache.iotdb.db.pipe.receiver.protocol.thrift.IoTDBDataNodeReceiverAgent;
+import org.apache.iotdb.db.protocol.session.ClientSession;
+import org.apache.iotdb.db.protocol.session.SessionManager;
+import org.apache.iotdb.db.queryengine.plan.Coordinator;
 import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
 import org.apache.iotdb.rpc.TSStatusCode;
 import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
@@ -73,6 +76,9 @@ public void runMayThrow() throws Throwable {
 
     LOGGER.info("Pipe air gap receiver {} started. Socket: {}", receiverId, socket);
 
+    final ClientSession session = new ClientSession(socket);
+    SessionManager.getInstance().registerSession(session);
+
     try {
       while (!socket.isClosed()) {
         isELanguagePayload = false;
@@ -91,6 +97,8 @@ public void runMayThrow() throws Throwable {
       throw e;
     } finally {
       PipeDataNodeAgent.receiver().thrift().handleClientExit();
+      SessionManager.getInstance()
+          .closeSession(session, Coordinator.getInstance()::cleanupQueryExecution);
       socket.close();
     }
   }