Fix

Author: timoninmaxim

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

@@ -840,9 +840,12 @@ private Collection<ClusterNode> nodes() throws IgniteCheckedException {
     }
 
     /**
+     * Collects query data nodes matching specified {@code prj} and {@code part}.
+     *
      * @param cctx Cache context.
      * @param prj Projection (optional).
-     * @return Collection of data nodes in provided projection (if any).
+     * @param part Partition (optional).
+     * @return Collection of data nodes matching specified {@code prj} and {@code part}.
      * @throws IgniteCheckedException If partition number is invalid.
      */
     private static Collection<ClusterNode> nodes(final GridCacheContext<?, ?> cctx,
@@ -856,25 +859,17 @@ private static Collection<ClusterNode> nodes(final GridCacheContext<?, ?> cctx,
         if (prj == null && part == null)
             return affNodes;
 
-        if (part != null && part >= cctx.affinity().partitions())
-            throw new IgniteCheckedException("Invalid partition number: " + part);
-
-        final Set<ClusterNode> partHolders;
-
         if (part != null) {
-            if (cctx.config().getCacheMode() == CacheMode.PARTITIONED)
-                partHolders = Collections.singleton(cctx.affinity().primaryByPartition(part, topVer));
-            else
-                partHolders = new HashSet<>(cctx.topology().owners(part, topVer));
+            if (part >= cctx.affinity().partitions())
+                throw new IgniteCheckedException("Invalid partition number: " + part);
+
+            affNodes = cctx.topology().nodes(part, topVer);
         }
-        else
-            partHolders = Collections.emptySet();
 
         return F.view(affNodes, new P1<ClusterNode>() {
             @Override public boolean apply(ClusterNode n) {
                 return cctx.discovery().cacheAffinityNode(n, cctx.name()) &&
-                    (prj == null || prj.node(n.id()) != null) &&
-                    (part == null || partHolders.contains(n));
+                    (prj == null || prj.node(n.id()) != null);
             }
         });
     }

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

@@ -25,6 +25,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.ignite.IgniteCheckedException;
@@ -38,6 +39,7 @@
 import org.apache.ignite.internal.processors.cache.query.reducer.TextQueryReducer;
 import org.apache.ignite.internal.processors.cache.query.reducer.UnsortedCacheQueryReducer;
 import org.apache.ignite.internal.util.lang.GridPlainCallable;
+import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.U;
 
 import static org.apache.ignite.internal.processors.cache.query.GridCacheQueryType.INDEX;
@@ -60,7 +62,7 @@ public class GridCacheDistributedQueryFuture<K, V, R> extends GridCacheQueryFutu
     /** Count of streams that finish receiving remote pages. */
     private final AtomicInteger noRemotePagesStreamsCnt = new AtomicInteger();
 
-    /** Count down this latch when every node responses on initial cache query request. */
+    /** Count down this latch when every localNodeOrRemote responses on initial cache query request. */
     private final CountDownLatch firstPageLatch = new CountDownLatch(1);
 
     /** Set of nodes that deliver their first page. */
@@ -92,7 +94,7 @@ protected GridCacheDistributedQueryFuture(
         qryMgr = (GridCacheDistributedQueryManager<K, V>)ctx.queries();
 
         if (qry.query().partition() != null)
-            nodes = Collections.singletonList(node(nodes));
+            nodes = Collections.singletonList(cctx.isReplicated() ? localOrRemoteNode(nodes) : F.first(nodes));
 
         streams = new ConcurrentHashMap<>(nodes.size());
 
@@ -118,19 +120,24 @@ protected GridCacheDistributedQueryFuture(
     }
 
     /**
-     * @return Nodes for query execution.
+     * @return A local node if available, otherwise a random node from the given collection.
      */
-    private ClusterNode node(Collection<ClusterNode> nodes) {
-        ClusterNode rmtNode = null;
+    private static ClusterNode localOrRemoteNode(Collection<ClusterNode> nodes) {
+        int remoteNodeIdx = ThreadLocalRandom.current().nextInt(nodes.size());
+
+        ClusterNode remoteNode = null;
 
         for (ClusterNode node : nodes) {
             if (node.isLocal())
                 return node;
 
-            rmtNode = node;
+            if (remoteNodeIdx-- == 0)
+                remoteNode = node;
         }
 
-        return rmtNode;
+        assert remoteNode != null;
+
+        return remoteNode;
     }
 
     /** {@inheritDoc} */
@@ -150,7 +157,7 @@ private ClusterNode node(Collection<ClusterNode> nodes) {
         NodePageStream<R> stream = streams.get(nodeId);
 
         if (stream != null && stream.hasRemotePages())
-            onError(new ClusterTopologyCheckedException("Remote node has left topology: " + nodeId));
+            onError(new ClusterTopologyCheckedException("Remote localNodeOrRemote has left topology: " + nodeId));
     }
 
     /** {@inheritDoc} */
@@ -282,7 +289,7 @@ private void cancelPages(UUID nodeId) {
             GridCacheQueryRequest req = GridCacheQueryRequest.cancelRequest(cctx, reqId, fields());
 
             if (nodeId.equals(cctx.localNodeId())) {
-                // Process cancel query directly (without sending) for local node,
+                // Process cancel query directly (without sending) for local localNodeOrRemote,
                 cctx.closures().callLocalSafe(new GridPlainCallable<Object>() {
                     @Override public Object call() {
                         qryMgr.processQueryRequest(cctx.localNodeId(), req);
@@ -298,10 +305,10 @@ private void cancelPages(UUID nodeId) {
                 catch (IgniteCheckedException e) {
                     if (cctx.io().checkNodeLeft(nodeId, e, false)) {
                         if (log.isDebugEnabled())
-                            log.debug("Failed to send cancel request, node failed: " + nodeId);
+                            log.debug("Failed to send cancel request, localNodeOrRemote failed: " + nodeId);
                     }
                     else
-                        U.error(log, "Failed to send cancel request [node=" + nodeId + ']', e);
+                        U.error(log, "Failed to send cancel request [localNodeOrRemote=" + nodeId + ']', e);
                 }
             }
         }