Avoid non-informative warning on query retry
@@ -130,7 +130,7 @@ public class GridReduceQueryExecutor {
private IgniteLogger log;
/** */
- private final AtomicLong qryIdGen = new AtomicLong();
+ private final AtomicLong qryReqIdGen = new AtomicLong();
/** */
private final ConcurrentMap<Long, ReduceQueryRun> runs = new ConcurrentHashMap<>();
@@ -419,11 +419,10 @@ public class GridReduceQueryExecutor {
H2PooledConnection conn = h2.connections().connection(schemaName);
- final long qryReqId = qryIdGen.incrementAndGet();
+ final long qryReqId = qryReqIdGen.incrementAndGet();
h2.runningQueryManager().trackRequestId(qryReqId);
- boolean retry = false;
boolean release = true;
try {
@@ -456,6 +455,8 @@ public class GridReduceQueryExecutor {
final C2<ClusterNode, Message, Message> spec =
parts == null ? null : new ReducePartitionsSpecializer(mapping.queryPartitionsMap());
+ boolean retry = false;
+
if (send(nodes, req, spec, false)) {
awaitAllReplies(r, nodes, cancel);
@@ -465,30 +466,29 @@ public class GridReduceQueryExecutor {
if (err != null) {
if (err.getCause() instanceof IgniteClientDisconnectedException)
throw err;
-
- if (QueryUtils.wasCancelled(err))
+ else if (QueryUtils.wasCancelled(err))
throw new QueryCancelledException(); // Throw correct exception.
throw err;
}
- else {
- retry = true;
// If remote node asks us to retry then we have outdated full partition map.
h2.awaitForReadyTopologyVersion(r.retryTopologyVersion());
+
+ retry = true;
}
}
- }
- else // Send failed.
+ else
retry = true;
if (retry) {
- lastRun = runs.remove(qryReqId);
+ lastRun = runs.get(qryReqId);
+
assert lastRun != null;
- continue;
+ continue; // Retry.
}
- else {
+
Iterator<List<?>> resIter;
if (skipMergeTbl) {
@@ -513,8 +513,7 @@ public class GridReduceQueryExecutor {
null,
null,
null,
- true
- );
+ true);
H2Utils.setupConnection(conn, qctx, false, enforceJoinOrder);
@@ -557,7 +556,6 @@ public class GridReduceQueryExecutor {
return new GridQueryCacheObjectsIterator(resIter, h2.objectContext(), keepBinary);
}
- }
catch (IgniteCheckedException | RuntimeException e) {
release = true;
@@ -593,7 +591,7 @@ public class GridReduceQueryExecutor {
}
}
finally {
- if (conn != null && (retry || release))
+ if (conn != null && release)
U.close(conn, log);
}
}
@@ -881,7 +879,7 @@ public class GridReduceQueryExecutor {
}
}
- final long reqId = qryIdGen.incrementAndGet();
+ final long reqId = qryReqIdGen.incrementAndGet();
h2.runningQueryManager().trackRequestId(reqId);
@@ -983,6 +981,7 @@ public class GridReduceQueryExecutor {
*/
void releaseRemoteResources(Collection<ClusterNode> nodes, ReduceQueryRun r, long qryReqId,
boolean distributedJoins, MvccQueryTracker mvccTracker) {
+ try {
if (distributedJoins)
send(nodes, new GridQueryCancelRequest(qryReqId), null, true);
@@ -997,12 +996,14 @@ public class GridReduceQueryExecutor {
break;
}
}
-
+ }
+ finally {
if (!runs.remove(qryReqId, r))
U.warn(log, "Query run was already removed: " + qryReqId);
else if (mvccTracker != null)
mvccTracker.onDone();
}
+ }
/**
* @param r Query run.
