must

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java

@@ -369,10 +369,10 @@ private void doTransfer(final PipeStatementInsertionEvent pipeStatementInsertion
   @Override
   public void close() throws Exception {
     if (session != null) {
-      SESSION_MANAGER.closeSession(session, COORDINATOR::cleanupQueryExecution);
+      SESSION_MANAGER.closeSession(session, COORDINATOR::cleanupQueryExecution, false);
     }
     if (treeSession != null) {
-      SESSION_MANAGER.closeSession(treeSession, COORDINATOR::cleanupQueryExecution);
+      SESSION_MANAGER.closeSession(treeSession, COORDINATOR::cleanupQueryExecution, false);
     }
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/SessionManager.java

@@ -255,6 +255,11 @@ public BasicOpenSessionResp login(
   }
 
   public boolean closeSession(IClientSession session, LongConsumer releaseByQueryId) {
+    return closeSession(session, releaseByQueryId, true);
+  }
+
+  public boolean closeSession(
+      IClientSession session, LongConsumer releaseByQueryId, boolean mustCurrent) {
     releaseSessionResource(session, releaseByQueryId);
     MetricService.getInstance()
         .remove(
@@ -264,11 +269,11 @@ public boolean closeSession(IClientSession session, LongConsumer releaseByQueryI
             String.valueOf(session.getId()));
     // TODO we only need to do so when query is killed by time out  close the socket.
     IClientSession session1 = currSession.get();
-    if (session1 != null && session != session1) {
+    if (mustCurrent && session1 != null && session != session1) {
       LOGGER.info(
           String.format(
               "The client-%s is trying to close another session %s, pls check if it's a bug",
-              session, session1));
+              session1, session));
       return false;
     } else {
       LOGGER.info(String.format("Session-%s is closing", session));