You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2019/04/18 17:46:11 UTC
[ignite] branch master updated: IGNITE-11489: SQL: Merge IO message
listener for GridTopic.TOPIC_QUERY into single listener. This closes #6405.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 28b69fd IGNITE-11489: SQL: Merge IO message listener for GridTopic.TOPIC_QUERY into single listener. This closes #6405.
28b69fd is described below
commit 28b69fd57e49c96f4f0a5854a807298d5c14f4e6
Author: Andrey V. Mashenkov <an...@gmail.com>
AuthorDate: Thu Apr 18 20:45:34 2019 +0300
IGNITE-11489: SQL: Merge IO message listener for GridTopic.TOPIC_QUERY into single listener. This closes #6405.
---
.../processors/query/h2/IgniteH2Indexing.java | 85 +++++++++++++-
.../query/h2/twostep/GridMapQueryExecutor.java | 106 ++++-------------
.../query/h2/twostep/GridReduceQueryExecutor.java | 110 ++++--------------
.../h2/twostep/RetryCauseMessageSelfTest.java | 126 ++++++++-------------
4 files changed, 174 insertions(+), 253 deletions(-)
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
index 369f045..bcca56f 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
@@ -44,11 +44,15 @@ import org.apache.ignite.cache.query.QueryCancelledException;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cache.query.SqlQuery;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.events.DiscoveryEvent;
+import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.GridTopic;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
import org.apache.ignite.internal.managers.IgniteMBeansManager;
+import org.apache.ignite.internal.managers.communication.GridMessageListener;
+import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.mxbean.SqlQueryMXBean;
import org.apache.ignite.internal.mxbean.SqlQueryMXBeanImpl;
import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
@@ -118,6 +122,12 @@ import org.apache.ignite.internal.processors.query.h2.sql.GridSqlStatement;
import org.apache.ignite.internal.processors.query.h2.twostep.GridMapQueryExecutor;
import org.apache.ignite.internal.processors.query.h2.twostep.GridReduceQueryExecutor;
import org.apache.ignite.internal.processors.query.h2.twostep.PartitionReservationManager;
+import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryCancelRequest;
+import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryFailResponse;
+import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryNextPageRequest;
+import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryNextPageResponse;
+import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlRequest;
+import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlResponse;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitor;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
@@ -265,6 +275,12 @@ public class IgniteH2Indexing implements GridQueryIndexing {
/** H2 Connection manager. */
private LongRunningQueryManager longRunningQryMgr;
+ /** Discovery event listener. */
+ private GridLocalEventListener discoLsnr;
+
+ /** Query message listener. */
+ private GridMessageListener qryLsnr;
+
/**
* @return Kernal context.
*/
@@ -1905,12 +1921,23 @@ public class IgniteH2Indexing implements GridQueryIndexing {
nodeId = ctx.localNodeId();
marshaller = ctx.config().getMarshaller();
- mapQryExec = new GridMapQueryExecutor(busyLock);
- rdcQryExec = new GridReduceQueryExecutor(busyLock);
+ mapQryExec = new GridMapQueryExecutor();
+ rdcQryExec = new GridReduceQueryExecutor();
mapQryExec.start(ctx, this);
rdcQryExec.start(ctx, this);
+ discoLsnr = evt -> {
+ mapQryExec.onNodeLeft((DiscoveryEvent)evt);
+ rdcQryExec.onNodeLeft((DiscoveryEvent)evt);
+ };
+
+ ctx.event().addLocalEventListener(discoLsnr, EventType.EVT_NODE_FAILED, EventType.EVT_NODE_LEFT);
+
+ qryLsnr = (nodeId, msg, plc) -> onMessage(nodeId, msg);
+
+ ctx.io().addMessageListener(GridTopic.TOPIC_QUERY, qryLsnr);
+
runningQryMgr = new RunningQueryManager(ctx);
partExtractor = new PartitionExtractor(new H2PartitionResolver(this), ctx);
@@ -1924,6 +1951,57 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
/**
+ * @param nodeId Node ID.
+ * @param msg Message.
+ */
+ public void onMessage(UUID nodeId, Object msg) {
+ assert msg != null;
+
+ ClusterNode node = ctx.discovery().node(nodeId);
+
+ if (node == null)
+ return; // Node left, ignore.
+
+ if (!busyLock.enterBusy())
+ return;
+
+ try {
+ if (msg instanceof GridCacheQueryMarshallable)
+ ((GridCacheQueryMarshallable)msg).unmarshall(ctx.config().getMarshaller(), ctx);
+
+ try {
+ boolean processed = true;
+
+ if (msg instanceof GridQueryNextPageRequest)
+ mapQueryExecutor().onNextPageRequest(node, (GridQueryNextPageRequest)msg);
+ else if (msg instanceof GridQueryNextPageResponse)
+ reduceQueryExecutor().onNextPage(node, (GridQueryNextPageResponse)msg);
+ else if (msg instanceof GridH2QueryRequest)
+ mapQueryExecutor().onQueryRequest(node, (GridH2QueryRequest)msg);
+ else if (msg instanceof GridH2DmlRequest)
+ mapQueryExecutor().onDmlRequest(node, (GridH2DmlRequest)msg);
+ else if (msg instanceof GridH2DmlResponse)
+ reduceQueryExecutor().onDmlResponse(node, (GridH2DmlResponse)msg);
+ else if (msg instanceof GridQueryFailResponse)
+ reduceQueryExecutor().onFail(node, (GridQueryFailResponse)msg);
+ else if (msg instanceof GridQueryCancelRequest)
+ mapQueryExecutor().onCancel(node, (GridQueryCancelRequest)msg);
+ else
+ processed = false;
+
+ if (processed && log.isDebugEnabled())
+ log.debug("Processed message " + msg.getClass().getName() + ": " + nodeId + "->" + ctx.localNodeId() + " " + msg);
+ }
+ catch (Throwable th) {
+ U.error(log, "Failed to process message: " + msg, th);
+ }
+ }
+ finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ /**
* @return Value object context.
*/
public CacheObjectValueContext objectContext() {
@@ -2193,6 +2271,9 @@ public class IgniteH2Indexing implements GridQueryIndexing {
/** {@inheritDoc} */
@Override public void onKernalStop() {
connMgr.onKernalStop();
+
+ ctx.io().removeMessageListener(GridTopic.TOPIC_QUERY, qryLsnr);
+ ctx.event().removeLocalEventListener(discoLsnr);
}
/**
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
index a42ab12..f5c5b43 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
@@ -41,17 +41,12 @@ import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.CacheQueryExecutedEvent;
import org.apache.ignite.events.DiscoveryEvent;
-import org.apache.ignite.events.Event;
-import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.GridTopic;
-import org.apache.ignite.internal.managers.communication.GridMessageListener;
-import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.query.CacheQueryType;
-import org.apache.ignite.internal.processors.cache.query.GridCacheQueryMarshallable;
import org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery;
import org.apache.ignite.internal.processors.query.GridQueryCancel;
import org.apache.ignite.internal.processors.query.h2.H2ConnectionWrapper;
@@ -70,7 +65,6 @@ import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQuery
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlRequest;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlResponse;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest;
-import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -107,16 +101,6 @@ public class GridMapQueryExecutor {
/** */
private ConcurrentMap<UUID, MapNodeResults> qryRess = new ConcurrentHashMap<>();
- /** */
- private final GridSpinBusyLock busyLock;
-
- /**
- * @param busyLock Busy lock.
- */
- public GridMapQueryExecutor(GridSpinBusyLock busyLock) {
- this.busyLock = busyLock;
- }
-
/**
* @param ctx Context.
* @param h2 H2 Indexing.
@@ -129,39 +113,23 @@ public class GridMapQueryExecutor {
qryCtxRegistry = h2.queryContextRegistry();
log = ctx.log(GridMapQueryExecutor.class);
+ }
- ctx.event().addLocalEventListener(new GridLocalEventListener() {
- @Override public void onEvent(final Event evt) {
- UUID nodeId = ((DiscoveryEvent)evt).eventNode().id();
-
- qryCtxRegistry.clearSharedOnRemoteNodeStop(nodeId);
-
- MapNodeResults nodeRess = qryRess.remove(nodeId);
-
- if (nodeRess == null)
- return;
+ /**
+ * Node left event handling method..
+ * @param evt Discovery event.
+ */
+ public void onNodeLeft(DiscoveryEvent evt) {
+ UUID nodeId = evt.eventNode().id();
- nodeRess.cancelAll();
- }
- }, EventType.EVT_NODE_FAILED, EventType.EVT_NODE_LEFT);
+ qryCtxRegistry.clearSharedOnRemoteNodeStop(nodeId);
- ctx.io().addMessageListener(GridTopic.TOPIC_QUERY, new GridMessageListener() {
- @SuppressWarnings("deprecation")
- @Override public void onMessage(UUID nodeId, Object msg, byte plc) {
- if (!busyLock.enterBusy())
- return;
+ MapNodeResults nodeRess = qryRess.remove(nodeId);
- try {
- if (msg instanceof GridCacheQueryMarshallable)
- ((GridCacheQueryMarshallable)msg).unmarshall(ctx.config().getMarshaller(), ctx);
+ if (nodeRess == null)
+ return;
- GridMapQueryExecutor.this.onMessage(nodeId, msg);
- }
- finally {
- busyLock.leaveBusy();
- }
- }
- });
+ nodeRess.cancelAll();
}
/**
@@ -173,44 +141,10 @@ public class GridMapQueryExecutor {
}
/**
- * @param nodeId Node ID.
- * @param msg Message.
- */
- public void onMessage(UUID nodeId, Object msg) {
- try {
- assert msg != null;
-
- ClusterNode node = ctx.discovery().node(nodeId);
-
- if (node == null)
- return; // Node left, ignore.
-
- boolean processed = true;
-
- if (msg instanceof GridH2QueryRequest)
- onQueryRequest(node, (GridH2QueryRequest)msg);
- else if (msg instanceof GridQueryNextPageRequest)
- onNextPageRequest(node, (GridQueryNextPageRequest)msg);
- else if (msg instanceof GridQueryCancelRequest)
- onCancel(node, (GridQueryCancelRequest)msg);
- else if (msg instanceof GridH2DmlRequest)
- onDmlRequest(node, (GridH2DmlRequest)msg);
- else
- processed = false;
-
- if (processed && log.isDebugEnabled())
- log.debug("Processed request: " + nodeId + "->" + ctx.localNodeId() + " " + msg);
- }
- catch(Throwable th) {
- U.error(log, "Failed to process message: " + msg, th);
- }
- }
-
- /**
* @param node Node.
* @param msg Message.
*/
- private void onCancel(ClusterNode node, GridQueryCancelRequest msg) {
+ public void onCancel(ClusterNode node, GridQueryCancelRequest msg) {
long qryReqId = msg.queryRequestId();
MapNodeResults nodeRess = resultsForNode(node.id());
@@ -250,7 +184,7 @@ public class GridMapQueryExecutor {
* @param req Query request.
* @throws IgniteCheckedException On error.
*/
- private void onQueryRequest(final ClusterNode node, final GridH2QueryRequest req) throws IgniteCheckedException {
+ public void onQueryRequest(final ClusterNode node, final GridH2QueryRequest req) throws IgniteCheckedException {
int[] qryParts = req.queryPartitions();
final Map<UUID,int[]> partsMap = req.partitions();
@@ -618,7 +552,7 @@ public class GridMapQueryExecutor {
* @param node Node.
* @param req DML request.
*/
- private void onDmlRequest(final ClusterNode node, final GridH2DmlRequest req) {
+ public void onDmlRequest(final ClusterNode node, final GridH2DmlRequest req) {
int[] parts = req.queryPartitions();
List<Integer> cacheIds = req.caches();
@@ -736,7 +670,7 @@ public class GridMapQueryExecutor {
if (node.isLocal()) {
U.error(log, "Failed to run map query on local node.", err);
- h2.reduceQueryExecutor().onMessage(ctx.localNodeId(), msg);
+ h2.reduceQueryExecutor().onFail(node, msg);
}
else
ctx.io().sendToGridTopic(node, GridTopic.TOPIC_QUERY, msg, QUERY_POOL);
@@ -766,7 +700,7 @@ public class GridMapQueryExecutor {
log.debug("Sending: [localNodeId=" + ctx.localNodeId() + ", node=" + node.id() + ", msg=" + rsp + "]");
if (node.isLocal())
- h2.reduceQueryExecutor().onMessage(ctx.localNodeId(), rsp);
+ h2.reduceQueryExecutor().onDmlResponse(node, rsp);
else {
rsp.marshall(ctx.config().getMarshaller());
@@ -782,7 +716,7 @@ public class GridMapQueryExecutor {
* @param node Node.
* @param req Request.
*/
- private void onNextPageRequest(final ClusterNode node, final GridQueryNextPageRequest req) {
+ public void onNextPageRequest(final ClusterNode node, final GridQueryNextPageRequest req) {
long reqId = req.queryRequestId();
final MapNodeResults nodeRess = qryRess.get(node.id());
@@ -934,7 +868,7 @@ public class GridMapQueryExecutor {
try {
if (msg != null) {
if (node.isLocal())
- h2.reduceQueryExecutor().onMessage(ctx.localNodeId(), msg);
+ h2.reduceQueryExecutor().onNextPage(node, msg);
else
ctx.io().sendToGridTopic(node, GridTopic.TOPIC_QUERY, msg, QUERY_POOL);
}
@@ -966,7 +900,7 @@ public class GridMapQueryExecutor {
msg.retryCause(retryCause);
if (loc)
- h2.reduceQueryExecutor().onMessage(ctx.localNodeId(), msg);
+ h2.reduceQueryExecutor().onNextPage(node, msg);
else
ctx.io().sendToGridTopic(node, GridTopic.TOPIC_QUERY, msg, QUERY_POOL);
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
index 3291295..f4e842b 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
@@ -46,18 +46,13 @@ import org.apache.ignite.cache.query.QueryCancelledException;
import org.apache.ignite.cache.query.QueryRetryException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.DiscoveryEvent;
-import org.apache.ignite.events.Event;
-import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.GridTopic;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.managers.communication.GridIoPolicy;
-import org.apache.ignite.internal.managers.communication.GridMessageListener;
-import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker;
-import org.apache.ignite.internal.processors.cache.query.GridCacheQueryMarshallable;
import org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery;
import org.apache.ignite.internal.processors.cache.query.GridCacheTwoStepQuery;
import org.apache.ignite.internal.processors.query.GridQueryCacheObjectsIterator;
@@ -82,7 +77,6 @@ import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlReque
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlResponse;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest;
import org.apache.ignite.internal.transactions.IgniteTxAlreadyCompletedCheckedException;
-import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.internal.util.typedef.C2;
import org.apache.ignite.internal.util.typedef.CIX2;
import org.apache.ignite.internal.util.typedef.F;
@@ -149,12 +143,12 @@ public class GridReduceQueryExecutor {
private final Lock fakeTblsLock = new ReentrantLock();
/** */
- private final GridSpinBusyLock busyLock;
-
- /** */
private final CIX2<ClusterNode,Message> locNodeHnd = new CIX2<ClusterNode,Message>() {
@Override public void applyx(ClusterNode locNode, Message msg) {
- h2.mapQueryExecutor().onMessage(locNode.id(), msg);
+ assert msg instanceof GridQueryNextPageRequest || msg instanceof GridH2QueryRequest ||
+ msg instanceof GridH2DmlRequest || msg instanceof GridQueryCancelRequest : msg.getClass();
+
+ h2.onMessage(locNode.id(), msg);
}
};
@@ -165,15 +159,6 @@ public class GridReduceQueryExecutor {
private long dfltQueryTimeout;
/**
- * Constructor.
- *
- * @param busyLock Busy lock.
- */
- public GridReduceQueryExecutor(GridSpinBusyLock busyLock) {
- this.busyLock = busyLock;
- }
-
- /**
* @param ctx Context.
* @param h2 H2 Indexing.
* @throws IgniteCheckedException If failed.
@@ -187,44 +172,27 @@ public class GridReduceQueryExecutor {
log = ctx.log(GridReduceQueryExecutor.class);
mapper = new ReducePartitionMapper(ctx, log);
+ }
- ctx.io().addMessageListener(GridTopic.TOPIC_QUERY, new GridMessageListener() {
- @SuppressWarnings("deprecation")
- @Override public void onMessage(UUID nodeId, Object msg, byte plc) {
- if (!busyLock.enterBusy())
- return;
+ /**
+ * Node left event handling method..
+ * @param evt Discovery event.
+ */
+ public void onNodeLeft(DiscoveryEvent evt) {
+ UUID nodeId = evt.eventNode().id();
- try {
- if (msg instanceof GridCacheQueryMarshallable)
- ((GridCacheQueryMarshallable)msg).unmarshall(ctx.config().getMarshaller(), ctx);
+ for (ReduceQueryRun r : runs.values()) {
+ for (ReduceIndex idx : r.indexes()) {
+ if (idx.hasSource(nodeId)) {
+ handleNodeLeft(r, nodeId);
- GridReduceQueryExecutor.this.onMessage(nodeId, msg);
- }
- finally {
- busyLock.leaveBusy();
+ break;
}
}
- });
-
- ctx.event().addLocalEventListener(new GridLocalEventListener() {
- @Override public void onEvent(final Event evt) {
- UUID nodeId = ((DiscoveryEvent)evt).eventNode().id();
-
- for (ReduceQueryRun r : runs.values()) {
- for (ReduceIndex idx : r.indexes()) {
- if (idx.hasSource(nodeId)) {
- handleNodeLeft(r, nodeId);
-
- break;
- }
- }
- }
-
- for (DmlDistributedUpdateRun r : updRuns.values())
- r.handleNodeLeft(nodeId);
+ }
- }
- }, EventType.EVT_NODE_FAILED, EventType.EVT_NODE_LEFT);
+ for (DmlDistributedUpdateRun r : updRuns.values())
+ r.handleNodeLeft(nodeId);
}
/**
@@ -236,42 +204,10 @@ public class GridReduceQueryExecutor {
}
/**
- * @param nodeId Node ID.
- * @param msg Message.
- */
- public void onMessage(UUID nodeId, Object msg) {
- try {
- assert msg != null;
-
- ClusterNode node = ctx.discovery().node(nodeId);
-
- if (node == null)
- return; // Node left, ignore.
-
- boolean processed = true;
-
- if (msg instanceof GridQueryNextPageResponse)
- onNextPage(node, (GridQueryNextPageResponse)msg);
- else if (msg instanceof GridQueryFailResponse)
- onFail(node, (GridQueryFailResponse)msg);
- else if (msg instanceof GridH2DmlResponse)
- onDmlResponse(node, (GridH2DmlResponse)msg);
- else
- processed = false;
-
- if (processed && log.isDebugEnabled())
- log.debug("Processed response: " + nodeId + "->" + ctx.localNodeId() + " " + msg);
- }
- catch(Throwable th) {
- U.error(log, "Failed to process message: " + msg, th);
- }
- }
-
- /**
* @param node Node.
* @param msg Message.
*/
- private void onFail(ClusterNode node, GridQueryFailResponse msg) {
+ public void onFail(ClusterNode node, GridQueryFailResponse msg) {
ReduceQueryRun r = runs.get(msg.queryRequestId());
fail(r, node.id(), msg.error(), msg.failCode());
@@ -307,7 +243,7 @@ public class GridReduceQueryExecutor {
* @param node Node.
* @param msg Message.
*/
- private void onNextPage(final ClusterNode node, final GridQueryNextPageResponse msg) {
+ public void onNextPage(final ClusterNode node, final GridQueryNextPageResponse msg) {
final long qryReqId = msg.queryRequestId();
final int qry = msg.query();
final int seg = msg.segmentId();
@@ -340,7 +276,7 @@ public class GridReduceQueryExecutor {
(byte)setDataPageScanEnabled(0, r.isDataPageScanEnabled()));
if (node.isLocal())
- h2.mapQueryExecutor().onMessage(ctx.localNodeId(), msg0);
+ h2.mapQueryExecutor().onNextPageRequest(node, msg0);
else
ctx.io().sendToGridTopic(node, GridTopic.TOPIC_QUERY, msg0, GridIoPolicy.QUERY_POOL);
}
@@ -928,7 +864,7 @@ public class GridReduceQueryExecutor {
* @param node Node.
* @param msg Message.
*/
- private void onDmlResponse(final ClusterNode node, GridH2DmlResponse msg) {
+ public void onDmlResponse(final ClusterNode node, GridH2DmlResponse msg) {
try {
long reqId = msg.requestId();
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/RetryCauseMessageSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/RetryCauseMessageSelfTest.java
index e52c6d7..b2143fe 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/RetryCauseMessageSelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/RetryCauseMessageSelfTest.java
@@ -17,12 +17,12 @@
package org.apache.ignite.internal.processors.query.h2.twostep;
-import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import javax.cache.CacheException;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.query.SqlQuery;
@@ -38,7 +38,6 @@ import org.apache.ignite.internal.processors.query.GridQueryProcessor;
import org.apache.ignite.internal.processors.query.h2.H2Utils;
import org.apache.ignite.internal.processors.query.h2.IgniteH2Indexing;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest;
-import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
@@ -89,19 +88,14 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
GridMapQueryExecutor mapQryExec = GridTestUtils.getFieldValue(h2Idx, IgniteH2Indexing.class, "mapQryExec");
GridTestUtils.setFieldValue(h2Idx, IgniteH2Indexing.class, "mapQryExec",
- new MockGridMapQueryExecutor(null) {
- @Override public void onMessage(UUID nodeId, Object msg) {
- if (GridH2QueryRequest.class.isAssignableFrom(msg.getClass())) {
- GridH2QueryRequest qryReq = (GridH2QueryRequest)msg;
+ new MockGridMapQueryExecutor() {
+ @Override
+ public void onQueryRequest(ClusterNode node, GridH2QueryRequest qryReq) throws IgniteCheckedException {
+ qryReq.caches().add(Integer.MAX_VALUE);
- qryReq.caches().add(Integer.MAX_VALUE);
+ startedExecutor.onQueryRequest(node, qryReq);
- startedExecutor.onMessage(nodeId, msg);
-
- qryReq.caches().remove(qryReq.caches().size() - 1);
- }
- else
- startedExecutor.onMessage(nodeId, msg);
+ qryReq.caches().remove(qryReq.caches().size() - 1);
}
}.insertRealExecutor(mapQryExec));
@@ -133,21 +127,20 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
final ConcurrentMap<PartitionReservationKey, GridReservable> reservations = reservations(h2Idx);
GridTestUtils.setFieldValue(h2Idx, IgniteH2Indexing.class, "mapQryExec",
- new MockGridMapQueryExecutor(null) {
- @Override public void onMessage(UUID nodeId, Object msg) {
- if (GridH2QueryRequest.class.isAssignableFrom(msg.getClass())) {
- final PartitionReservationKey grpKey = new PartitionReservationKey(ORG, null);
+ new MockGridMapQueryExecutor() {
+ @Override
+ public void onQueryRequest(ClusterNode node, GridH2QueryRequest qryReq) throws IgniteCheckedException {
+ final PartitionReservationKey grpKey = new PartitionReservationKey(ORG, null);
- reservations.put(grpKey, new GridReservable() {
+ reservations.put(grpKey, new GridReservable() {
- @Override public boolean reserve() {
- return false;
- }
+ @Override public boolean reserve() {
+ return false;
+ }
- @Override public void release() {}
- });
- }
- startedExecutor.onMessage(nodeId, msg);
+ @Override public void release() {}
+ });
+ startedExecutor.onQueryRequest(node, qryReq);
}
}.insertRealExecutor(mapQryExec));
@@ -180,25 +173,19 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
final GridKernalContext ctx = GridTestUtils.getFieldValue(mapQryExec, GridMapQueryExecutor.class, "ctx");
GridTestUtils.setFieldValue(h2Idx, IgniteH2Indexing.class, "mapQryExec",
- new MockGridMapQueryExecutor(null) {
- @Override public void onMessage(UUID nodeId, Object msg) {
- if (GridH2QueryRequest.class.isAssignableFrom(msg.getClass())) {
- GridH2QueryRequest qryReq = (GridH2QueryRequest)msg;
-
- GridCacheContext<?, ?> cctx = ctx.cache().context().cacheContext(qryReq.caches().get(0));
+ new MockGridMapQueryExecutor() {
+ @Override public void onQueryRequest(ClusterNode node, GridH2QueryRequest qryReq) throws IgniteCheckedException {
+ GridCacheContext<?, ?> cctx = ctx.cache().context().cacheContext(qryReq.caches().get(0));
- GridDhtLocalPartition part = cctx.topology().localPartition(0, NONE, false);
+ GridDhtLocalPartition part = cctx.topology().localPartition(0, NONE, false);
- AtomicLong aState = GridTestUtils.getFieldValue(part, GridDhtLocalPartition.class, "state");
+ AtomicLong aState = GridTestUtils.getFieldValue(part, GridDhtLocalPartition.class, "state");
- long stateVal = aState.getAndSet(2);
+ long stateVal = aState.getAndSet(2);
- startedExecutor.onMessage(nodeId, msg);
+ startedExecutor.onQueryRequest(node, qryReq);
- aState.getAndSet(stateVal);
- }
- else
- startedExecutor.onMessage(nodeId, msg);
+ aState.getAndSet(stateVal);
}
}.insertRealExecutor(mapQryExec));
@@ -230,26 +217,20 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
final GridKernalContext ctx = GridTestUtils.getFieldValue(mapQryExec, GridMapQueryExecutor.class, "ctx");
GridTestUtils.setFieldValue(h2Idx, IgniteH2Indexing.class, "mapQryExec",
- new MockGridMapQueryExecutor(null) {
- @Override public void onMessage(UUID nodeId, Object msg) {
- if (GridH2QueryRequest.class.isAssignableFrom(msg.getClass())) {
- GridH2QueryRequest qryReq = (GridH2QueryRequest)msg;
-
- GridCacheContext<?, ?> cctx = ctx.cache().context().cacheContext(qryReq.caches().get(0));
+ new MockGridMapQueryExecutor() {
+ @Override
+ public void onQueryRequest(ClusterNode node, GridH2QueryRequest qryReq) throws IgniteCheckedException {
+ GridCacheContext<?, ?> cctx = ctx.cache().context().cacheContext(qryReq.caches().get(0));
- GridDhtLocalPartition part = cctx.topology().localPartition(0, NONE, false);
+ GridDhtLocalPartition part = cctx.topology().localPartition(0, NONE, false);
- AtomicLong aState = GridTestUtils.getFieldValue(part, GridDhtLocalPartition.class, "state");
+ AtomicLong aState = GridTestUtils.getFieldValue(part, GridDhtLocalPartition.class, "state");
- long stateVal = aState.getAndSet(2);
+ long stateVal = aState.getAndSet(2);
- startedExecutor.onMessage(nodeId, msg);
-
- aState.getAndSet(stateVal);
- }
- else
- startedExecutor.onMessage(nodeId, msg);
+ startedExecutor.onQueryRequest(node, qryReq);
+ aState.getAndSet(stateVal);
}
}.insertRealExecutor(mapQryExec));
@@ -281,24 +262,20 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
final ConcurrentMap<PartitionReservationKey, GridReservable> reservations = reservations(h2Idx);
GridTestUtils.setFieldValue(h2Idx, IgniteH2Indexing.class, "mapQryExec",
- new MockGridMapQueryExecutor(null) {
- @Override public void onMessage(UUID nodeId, Object msg) {
- if (GridH2QueryRequest.class.isAssignableFrom(msg.getClass())) {
- final PartitionReservationKey grpKey = new PartitionReservationKey(ORG, null);
-
- reservations.put(grpKey, new GridReservable() {
+ new MockGridMapQueryExecutor() {
+ @Override
+ public void onQueryRequest(ClusterNode node, GridH2QueryRequest qryReq) throws IgniteCheckedException {
+ final PartitionReservationKey grpKey = new PartitionReservationKey(ORG, null);
- @Override public boolean reserve() {
- throw H2Utils.retryException("test retry exception");
- }
+ reservations.put(grpKey, new GridReservable() {
+ @Override public boolean reserve() {
+ throw H2Utils.retryException("test retry exception");
+ }
- @Override public void release() {
- }
- });
- }
-
- startedExecutor.onMessage(nodeId, msg);
+ @Override public void release() {}
+ });
+ startedExecutor.onQueryRequest(node, qryReq);
}
}.insertRealExecutor(mapQryExec));
@@ -391,16 +368,9 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
return this;
}
- /**
- * @param busyLock Busy lock.
- */
- MockGridMapQueryExecutor(GridSpinBusyLock busyLock) {
- super(busyLock);
- }
-
/** {@inheritDoc} */
- @Override public void onMessage(UUID nodeId, Object msg) {
- startedExecutor.onMessage(nodeId, msg);
+ @Override public void onQueryRequest(ClusterNode node, GridH2QueryRequest req) throws IgniteCheckedException {
+ startedExecutor.onQueryRequest(node, req);
}
}
}