diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MergeJoinNode.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MergeJoinNode.java
index 800e66e918b72..57b8a488130b4 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MergeJoinNode.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MergeJoinNode.java
@@ -84,9 +84,6 @@ public abstract class MergeJoinNode<Row> extends AbstractNode<Row> {
      */
     protected final boolean distributed;
 
-    /** Flag indicating that join is in finishing stage (one of the inputs are ended, no more rows will be produced). */
-    protected boolean finishing;
-
     /**
      * @param ctx Execution context.
      * @param comp Join expression.
@@ -178,10 +175,10 @@ private void pushLeft(Row row) throws Exception {
 
         waitingLeft--;
 
-        if (!finishing)
-            leftInBuf.add(row);
+        leftInBuf.add(row);
 
-        join();
+        if (waitingLeft == 0 && waitingRight <= 0)
+            join();
     }
 
     /** */
@@ -191,10 +188,10 @@ private void pushRight(Row row) throws Exception {
 
         waitingRight--;
 
-        if (!finishing)
-            rightInBuf.add(row);
+        rightInBuf.add(row);
 
-        join();
+        if (waitingRight == 0 && waitingLeft <= 0)
+            join();
     }
 
     /** */
@@ -204,7 +201,8 @@ private void endLeft() throws Exception {
 
         waitingLeft = NOT_WAITING;
 
-        join();
+        if (waitingRight <= 0)
+            join();
     }
 
     /** */
@@ -214,7 +212,8 @@ private void endRight() throws Exception {
 
         waitingRight = NOT_WAITING;
 
-        join();
+        if (waitingLeft <= 0)
+            join();
     }
 
     /** */
@@ -238,27 +237,6 @@ protected boolean rightFinished(boolean withMaterialization) {
             && (!withMaterialization || rightMaterialization == null);
     }
 
-    /** */
-    protected boolean checkJoinFinished() throws Exception {
-        if (!finishing) {
-            finishing = true;
-            leftInBuf.clear();
-            rightInBuf.clear();
-            rightMaterialization = null;
-            rightIdx = 0;
-            drainMaterialization = false;
-        }
-
-        if (!distributed || (waitingLeft == NOT_WAITING && waitingRight == NOT_WAITING)) {
-            requested = 0;
-            downstream().end();
-
-            return true;
-        }
-
-        return false;
-    }
-
     /** */
     protected void tryToRequestInputs() throws Exception {
         if (waitingLeft == 0 && leftInBuf.size() <= HALF_BUF_SIZE)
@@ -414,8 +392,12 @@ else if (cmp > 0) {
                 inLoop = false;
             }
 
-            if (requested > 0 && (leftFinished() || rightFinished(true)) && checkJoinFinished())
+            if (requested > 0 && (leftFinished() || rightFinished(true))) {
+                requested = 0;
+                downstream().end();
+
                 return;
+            }
 
             tryToRequestInputs();
         }
@@ -568,8 +550,12 @@ else if (cmp > 0) {
                 inLoop = false;
             }
 
-            if (requested > 0 && leftFinished() && checkJoinFinished())
+            if (requested > 0 && leftFinished()) {
+                requested = 0;
+                downstream().end();
+
                 return;
+            }
 
             tryToRequestInputs();
         }
@@ -734,8 +720,12 @@ else if (cmp > 0) {
                 inLoop = false;
             }
 
-            if (requested > 0 && rightFinished(true) && checkJoinFinished())
+            if (requested > 0 && rightFinished(true)) {
+                requested = 0;
+                downstream().end();
+
                 return;
+            }
 
             tryToRequestInputs();
         }
@@ -939,8 +929,12 @@ else if (cmp > 0) {
                 inLoop = false;
             }
 
-            if (requested > 0 && leftFinished() && rightFinished(true) && checkJoinFinished())
+            if (requested > 0 && leftFinished() && rightFinished(true)) {
+                requested = 0;
+                downstream().end();
+
                 return;
+            }
 
             tryToRequestInputs();
         }
@@ -995,8 +989,12 @@ else if (cmp > 0) {
                 inLoop = false;
             }
 
-            if (requested > 0 && (leftFinished() || rightFinished(false)) && checkJoinFinished())
+            if (requested > 0 && (leftFinished() || rightFinished(false))) {
+                requested = 0;
+                downstream().end();
+
                 return;
+            }
 
             tryToRequestInputs();
         }
@@ -1054,8 +1052,12 @@ else if (cmp > 0) {
                 inLoop = false;
             }
 
-            if (requested > 0 && leftFinished() && checkJoinFinished())
+            if (requested > 0 && leftFinished()) {
+                requested = 0;
+                downstream().end();
+
                 return;
+            }
 
             tryToRequestInputs();
         }
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/JoinBuffersExecutionTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/JoinBuffersExecutionTest.java
index 4ff685131cea0..0d52216bd9836 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/JoinBuffersExecutionTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/JoinBuffersExecutionTest.java
@@ -34,8 +34,6 @@
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.testframework.GridTestUtils;
 import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
 
 import static org.apache.calcite.rel.core.JoinRelType.ANTI;
 import static org.apache.calcite.rel.core.JoinRelType.FULL;
@@ -45,7 +43,6 @@
 import static org.apache.calcite.rel.core.JoinRelType.SEMI;
 
 /** Tests that buffers of join nodes are cleared at the join end and that a join node is not stuck. */
-@RunWith(Parameterized.class)
 public class JoinBuffersExecutionTest extends AbstractExecutionTest {
     /** */
     @Test
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java
index bc7df8a544a23..61be928ac2c71 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java
@@ -105,6 +105,30 @@ public void testCorrelatesCollision() {
             .check();
     }
 
+    /** */
+    @Test
+    public void testMergeJoinUnderCorrelateSurvivesRewind() {
+        sql("CREATE TABLE t0 (a INTEGER, b INTEGER) WITH " + atomicity());
+        sql("CREATE TABLE t1 (a INTEGER, b INTEGER) WITH " + atomicity());
+        sql("CREATE TABLE t2 (a INTEGER, b INTEGER) WITH " + atomicity());
+
+        sql("INSERT INTO t0 VALUES (1, 1), (2, 2), (3, 3)");
+        sql("INSERT INTO t1 VALUES (1, 1), (2, 2), (3, 3)");
+        sql("INSERT INTO t2 VALUES (1, 1)");
+
+        String qry = "SELECT t0.a, (SELECT /*+ MERGE_JOIN */ count(*) FROM t1 LEFT JOIN t2 ON t1.a = t2.a " +
+            "WHERE t1.a = (SELECT t0.a)) FROM t0";
+
+        // The defect is only reachable while the plan keeps a merge join under the correlate.
+        assertQuery(qry)
+            .matches(QueryChecker.containsSubPlan("IgniteCorrelatedNestedLoopJoin"))
+            .matches(QueryChecker.containsSubPlan("IgniteMergeJoin"))
+            .returns(1, 1L)
+            .returns(2, 1L)
+            .returns(3, 1L)
+            .check();
+    }
+
     /**
      * Tests colocated join possible with the help of correlated distribution.
      */