diff --git a/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/query/impl/join/JoinHandler.java b/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/query/impl/join/JoinHandler.java
index d5c99d33c..2340f7167 100644
--- a/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/query/impl/join/JoinHandler.java
+++ b/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/query/impl/join/JoinHandler.java
@@ -50,6 +50,8 @@ public class JoinHandler extends OwnThreadDMLHandler {
     protected List<FieldPacket> leftFieldPackets;
     protected List<FieldPacket> rightFieldPackets;
     private AtomicBoolean fieldSent = new AtomicBoolean(false);
+    // merge loop ended; set before terminateHandlerTree so late backend rows skip putLast on a full queue
+    private final AtomicBoolean joinMergeFinished = new AtomicBoolean(false);
     private BufferPool pool;
     private RowDataComparator leftComparator;
     private RowDataComparator rightComparator;
@@ -122,13 +124,17 @@ private Item makeOtherJoinOnItem(List<FieldPacket> rowPackets, BackendConnection
     @Override
     public boolean rowResponse(byte[] rowNull, RowDataPacket rowPacket, boolean isLeft, BackendConnection conn) {
         LOGGER.debug("rowresponse");
-        if (terminate.get()) {
+        // return true: interrupted; do not block on putLast after merge or during terminate
+        if (terminate.get() || joinMergeFinished.get()) {
             return true;
         }
         try {
             if (isLeft) {
                 leftLock.lock();
                 try {
+                    if (terminate.get() || joinMergeFinished.get()) {
+                        return true;
+                    }
                     addRowToDeque(rowPacket, leftFieldPackets.size(), leftQueue, leftComparator);
                 } finally {
                     leftLock.unlock();
@@ -136,6 +142,9 @@ public boolean rowResponse(byte[] rowNull, RowDataPacket rowPacket, boolean isLe
             } else {
                 rightLock.lock();
                 try {
+                    if (terminate.get() || joinMergeFinished.get()) {
+                        return true;
+                    }
                     addRowToDeque(rowPacket, rightFieldPackets.size(), rightQueue, rightComparator);
                 } finally {
                     rightLock.unlock();
@@ -151,7 +160,7 @@ public boolean rowResponse(byte[] rowNull, RowDataPacket rowPacket, boolean isLe
     @Override
     public void rowEofResponse(byte[] data, boolean isLeft, BackendConnection conn) {
         LOGGER.debug("roweof");
-        if (terminate.get()) {
+        if (terminate.get() || joinMergeFinished.get()) {
             return;
         }
         RowDataPacket eofRow = TERMINATED_ROW;
@@ -219,6 +228,9 @@ protected void ownThreadJob(Object... objects) {
                     rightLocal = takeFirst(rightQueue);
                 }
             }
+            // stop accepting backend rows and drain queues before waiting on connections in terminateHandlerTree
+            joinMergeFinished.set(true);
+            clearJoinQueuesAfterMerge();
 
             HandlerTool.terminateHandlerTree(this);
             // for trace, when join end before all rows return ,the handler should mark as finished
@@ -243,6 +255,15 @@ protected void ownThreadJob(Object... objects) {
         }
     }
 
+    /**
+     * Clear join queues after merge. Unlocked poll first so producers blocked on a full
+     * deque can finish;
+     */
+    private void clearJoinQueuesAfterMerge() {
+        clearDeque(leftQueue);
+        clearDeque(rightQueue);
+    }
+
     private LocalResult takeFirst(FairLinkedBlockingDeque<LocalResult> deque) throws InterruptedException {
         /**
          * it must be in single thread
diff --git a/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/query/impl/join/NotInHandler.java b/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/query/impl/join/NotInHandler.java
index 76f45439e..a51794bd8 100644
--- a/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/query/impl/join/NotInHandler.java
+++ b/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/query/impl/join/NotInHandler.java
@@ -29,6 +29,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantLock;
 
 public class NotInHandler extends OwnThreadDMLHandler {
     private static final Logger LOGGER = LoggerFactory.getLogger(NotInHandler.class);
@@ -43,7 +44,12 @@ public class NotInHandler extends OwnThreadDMLHandler {
     private RowDataComparator leftComparator;
     private RowDataComparator rightComparator;
     private AtomicBoolean fieldSent = new AtomicBoolean(false);
+    // merge loop ended; set before terminateHandlerTree so late backend rows skip putLast on a full queue
+    private final AtomicBoolean joinMergeFinished = new AtomicBoolean(false);
     private String charset = "UTF-8";
+    // prevent multi thread rowresponse
+    protected ReentrantLock leftLock = new ReentrantLock();
+    protected ReentrantLock rightLock = new ReentrantLock();
 
     public NotInHandler(long id, NonBlockingSession session, List<Order> leftOrder, List<Order> rightOrder) {
         super(id, session);
@@ -88,14 +94,31 @@ public void fieldEofResponse(byte[] headerNull, List<byte[]> fieldsNull, final L
     @Override
     public boolean rowResponse(byte[] rowNull, RowDataPacket rowPacket, boolean isLeft, BackendConnection conn) {
         LOGGER.debug("rowresponse");
-        if (terminate.get()) {
+        // return true: interrupted; do not block on putLast after merge or during terminate
+        if (terminate.get() || joinMergeFinished.get()) {
             return true;
         }
         try {
             if (isLeft) {
-                addRowToDeque(rowPacket, leftFieldPackets.size(), leftQueue, leftComparator);
+                leftLock.lock();
+                try {
+                    if (terminate.get() || joinMergeFinished.get()) {
+                        return true;
+                    }
+                    addRowToDeque(rowPacket, leftFieldPackets.size(), leftQueue, leftComparator);
+                } finally {
+                    leftLock.unlock();
+                }
             } else {
-                addRowToDeque(rowPacket, rightFieldPackets.size(), rightQueue, rightComparator);
+                rightLock.lock();
+                try {
+                    if (terminate.get() || joinMergeFinished.get()) {
+                        return true;
+                    }
+                    addRowToDeque(rowPacket, rightFieldPackets.size(), rightQueue, rightComparator);
+                } finally {
+                    rightLock.unlock();
+                }
             }
         } catch (InterruptedException e) {
             LOGGER.info("not in row exception", e);
@@ -107,7 +130,7 @@ public boolean rowResponse(byte[] rowNull, RowDataPacket rowPacket, boolean isLe
     @Override
     public void rowEofResponse(byte[] data, boolean isLeft, BackendConnection conn) {
         LOGGER.info("roweof");
-        if (terminate.get()) {
+        if (terminate.get() || joinMergeFinished.get()) {
             return;
         }
         RowDataPacket eofRow = TERMINATED_ROW;
@@ -164,6 +187,9 @@ protected void ownThreadJob(Object... objects) {
                     rightLocal = takeFirst(rightQueue);
                 }
             }
+            // stop accepting backend rows and drain queues before waiting on connections in terminateHandlerTree
+            joinMergeFinished.set(true);
+            clearJoinQueuesAfterMerge();
             session.setHandlerEnd(this);
             nextHandler.rowEofResponse(null, isLeft, conn);
             HandlerTool.terminateHandlerTree(this);
@@ -179,6 +205,15 @@ protected void ownThreadJob(Object... objects) {
         }
     }
 
+    /**
+     * Clear join queues after merge. Unlocked poll first so producers blocked on a full
+     * deque can finish;
+     */
+    private void clearJoinQueuesAfterMerge() {
+        clearDeque(leftQueue);
+        clearDeque(rightQueue);
+    }
+
     private LocalResult takeFirst(FairLinkedBlockingDeque<LocalResult> deque) throws InterruptedException {
         deque.waitUtilCount(1);
         LocalResult result = deque.peekFirst();