Fix code and add checkRequestFutureMapEmpty to test

Author: korot-7

modules/core/src/main/java/org/apache/ignite/internal/QueryMXBeanImpl.java

@@ -176,18 +176,10 @@ private static class CancelScan implements IgniteClosure<T3<UUID, String, Long>,
                 return null;
             }
 
-            GridCacheQueryManager qryMgr = ctx.queries();
+            GridCacheQueryRequest cancelReq = GridCacheQueryRequest.cancelRequest(ctx, arg.get3(), false);
 
-            if (qryMgr instanceof GridCacheDistributedQueryManager) {
-                GridCacheDistributedQueryManager distQryMgr = (GridCacheDistributedQueryManager) qryMgr;
-
-                GridCacheQueryRequest cancelReq = GridCacheQueryRequest.cancelRequest(ctx, arg.get3(), false);
-
-                distQryMgr.processQueryRequest(arg.get1(), cancelReq);
-
-            } else {
-                qryMgr.removeQueryResult(arg.get1(), arg.get3());
-            }
+            GridCacheDistributedQueryManager distQryMgr = (GridCacheDistributedQueryManager) ctx.queries();
+            distQryMgr.processQueryRequest(arg.get1(), cancelReq);
 
             return null;
         }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java

@@ -27,6 +27,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.query.QueryCancelledException;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.events.DiscoveryEvent;
 import org.apache.ignite.events.Event;
@@ -231,7 +232,7 @@ protected void removeQueryFuture(long reqId) {
                 GridCacheQueryResponse closedResponse = new GridCacheQueryResponse(
                         cctx.cacheId(),
                         req.id(),
-                        new NoSuchElementException("Iterator has been closed."),
+                        new QueryCancelledException("Iterator has been closed."),
                         cctx.deploymentEnabled()
                 );
 

modules/indexing/src/test/java/org/apache/ignite/util/KillCommandsTests.java

@@ -17,12 +17,7 @@
 
 package org.apache.ignite.util;
 
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import java.util.UUID;
+import java.util.*;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -157,12 +152,6 @@ public static void checkScanQueryCancelBeforeFetching(
         // Cancel first query.
         qryCanceler.accept(qryInfo);
 
-        // Checking that second query works fine after canceling first.
-        for (int i = 0; i < PAGE_SZ * PAGES_CNT - 1; i++)
-            assertNotNull(iter2.next());
-
-        qry2.close();
-
         // Fetch of the next page should throw the exception. New page is delivered in parallel to iterating.
         try {
             for (int i = 0; i < PAGE_SZ * PAGES_CNT - 1; i++)
@@ -180,7 +169,15 @@ public static void checkScanQueryCancelBeforeFetching(
             }
         }
 
+        // Checking that second query works fine after canceling first.
+        for (int i = 0; i < PAGE_SZ * PAGES_CNT - 1; i++)
+            assertNotNull(iter2.next());
+
         checkScanQueryResources(cli, srvs, qryInfo.get3());
+
+        qry2.close();
+
+        checkRequestFutureMapEmpty(cli, srvs, qryInfo.get1());
     }
 
     /**
@@ -259,6 +256,41 @@ private static void checkScanQueryResources(IgniteEx cli, List<IgniteEx> srvs, l
         }
     }
 
+
+    /**
+     * Checks that RequestFutureMap is empty on all nodes after query cancellation.
+     *
+     * @param cli Client node.
+     * @param srvs Server nodes.
+     * @param originNodeId Origin node ID.
+     */
+    private static void checkRequestFutureMapEmpty(
+            IgniteEx cli,
+            List<IgniteEx> srvs,
+            UUID originNodeId
+    ) {
+        List<IgniteEx> allNodes = new ArrayList<>(srvs);
+        allNodes.add(cli);
+
+        for (IgniteEx node : allNodes) {
+            int cacheId = CU.cacheId(DEFAULT_CACHE_NAME);
+            GridCacheContext<?, ?> ctx = node.context().cache().context().cacheContext(cacheId);
+
+            if (ctx == null) {
+                continue;
+            }
+
+            GridCacheQueryManager<?, ?> qryMgr = ctx.queries();
+
+            Map<UUID, GridCacheQueryManager.RequestFutureMap> qryIters = GridTestUtils.getFieldValue(qryMgr, "qryIters");
+            GridCacheQueryManager.RequestFutureMap futMap = qryIters.get(originNodeId);
+
+            if (futMap != null) {
+                assertTrue("RequestFutureMap not empty on node " + node.localNode().id() + ": size = " + futMap.size(), futMap.isEmpty());
+            }
+        }
+    }
+
     /**
      * Gets scan query info.
      *