You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2018/08/27 12:35:48 UTC
ignite git commit: IGNITE-9147 Fixed race between async tx rollback
and lock mapping on near node - Fixes #4574.
Repository: ignite
Updated Branches:
refs/heads/master b6182ab68 -> 3882ec048
IGNITE-9147 Fixed race between async tx rollback and lock mapping on near node - Fixes #4574.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/3882ec04
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/3882ec04
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/3882ec04
Branch: refs/heads/master
Commit: 3882ec0488b7be1fb6e74e14ecbeb9b396fd932f
Parents: b6182ab
Author: Aleksei Scherbakov <al...@gmail.com>
Authored: Mon Aug 27 15:34:22 2018 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Mon Aug 27 15:34:22 2018 +0300
----------------------------------------------------------------------
.../ignite/internal/TransactionsMXBeanImpl.java | 20 +-
.../internal/commandline/CommandHandler.java | 24 +-
.../colocated/GridDhtColocatedLockFuture.java | 358 +++++++++++--------
.../GridNearPessimisticTxPrepareFuture.java | 1 -
.../near/GridNearTxFinishFuture.java | 12 +-
.../ignite/internal/visor/tx/VisorTxInfo.java | 58 ++-
.../ignite/internal/visor/tx/VisorTxTask.java | 20 +-
.../cache/transactions/TxRollbackAsyncTest.java | 145 +++++++-
.../transactions/TxRollbackOnTimeoutTest.java | 26 +-
.../TxRollbackOnTopologyChangeTest.java | 16 -
.../junits/common/GridCommonAbstractTest.java | 23 ++
.../ignite/util/GridCommandHandlerTest.java | 23 --
12 files changed, 457 insertions(+), 269 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/main/java/org/apache/ignite/internal/TransactionsMXBeanImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/TransactionsMXBeanImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/TransactionsMXBeanImpl.java
index 210a320..16738de 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/TransactionsMXBeanImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/TransactionsMXBeanImpl.java
@@ -22,13 +22,10 @@ import java.io.StringWriter;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
import java.util.stream.Collectors;
import org.apache.ignite.IgniteCompute;
import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.visor.VisorTaskArgument;
import org.apache.ignite.internal.visor.tx.VisorTxInfo;
import org.apache.ignite.internal.visor.tx.VisorTxOperation;
@@ -37,7 +34,6 @@ import org.apache.ignite.internal.visor.tx.VisorTxSortOrder;
import org.apache.ignite.internal.visor.tx.VisorTxTask;
import org.apache.ignite.internal.visor.tx.VisorTxTaskArg;
import org.apache.ignite.internal.visor.tx.VisorTxTaskResult;
-import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.mxbean.TransactionsMXBean;
/**
@@ -103,21 +99,7 @@ public class TransactionsMXBeanImpl implements TransactionsMXBean {
w.println(key.toString());
for (VisorTxInfo info : entry.getValue().getInfos())
- w.println(" Tx: [xid=" + info.getXid() +
- ", label=" + info.getLabel() +
- ", state=" + info.getState() +
- ", startTime=" + info.getFormattedStartTime() +
- ", duration=" + info.getDuration() / 1000 +
- ", isolation=" + info.getIsolation() +
- ", concurrency=" + info.getConcurrency() +
- ", timeout=" + info.getTimeout() +
- ", size=" + info.getSize() +
- ", dhtNodes=" + F.transform(info.getPrimaryNodes(), new IgniteClosure<UUID, String>() {
- @Override public String apply(UUID id) {
- return U.id8(id);
- }
- }) +
- ']');
+ w.println(info.toUserString());
}
w.flush();
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/main/java/org/apache/ignite/internal/commandline/CommandHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/commandline/CommandHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/commandline/CommandHandler.java
index 7b5ce44..092efff 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/commandline/CommandHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/commandline/CommandHandler.java
@@ -1060,29 +1060,7 @@ public class CommandHandler {
"]");
for (VisorTxInfo info : entry.getValue().getInfos())
- log(" Tx: [xid=" + info.getXid() +
- ", label=" + info.getLabel() +
- ", state=" + info.getState() +
- ", startTime=" + info.getFormattedStartTime() +
- ", duration=" + info.getDuration() / 1000 +
- ", isolation=" + info.getIsolation() +
- ", concurrency=" + info.getConcurrency() +
- ", timeout=" + info.getTimeout() +
- ", size=" + info.getSize() +
- ", dhtNodes=" + (info.getPrimaryNodes() == null ? "N/A" :
- F.transform(info.getPrimaryNodes(), new IgniteClosure<UUID, String>() {
- @Override public String apply(UUID id) {
- return U.id8(id);
- }
- })) +
- ", nearXid=" + info.getNearXid() +
- ", parentNodeIds=" + (info.getMasterNodeIds() == null ? "N/A" :
- F.transform(info.getMasterNodeIds(), new IgniteClosure<UUID, String>() {
- @Override public String apply(UUID id) {
- return U.id8(id);
- }
- })) +
- ']');
+ log(info.toUserString());
}
}
catch (Throwable e) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
index a5ff2f3..27b3667 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
@@ -23,7 +23,6 @@ import java.util.Collection;
import java.util.Deque;
import java.util.HashSet;
import java.util.Iterator;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
@@ -180,6 +179,9 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
/** */
private int miniId;
+ /** {@code True} when mappings are ready for processing. */
+ private boolean mappingsReady;
+
/**
* @param cctx Registry.
* @param keys Keys to lock.
@@ -574,9 +576,24 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
* Cancellation has special meaning for lock futures. It's called then lock must be released on rollback.
*/
@Override public boolean cancel() {
- if (inTx())
+ if (inTx()) {
onError(tx.rollbackException());
+ /** Should wait until {@link mappings} are ready before continuing with async rollback
+ * or some primary nodes might not receive tx finish messages because of race.
+ * If prepare phase has not started waiting is not necessary.
+ */
+ synchronized (this) {
+ while (!mappingsReady)
+ try {
+ wait();
+ }
+ catch (InterruptedException e) {
+ // Ignore interrupts.
+ }
+ }
+ }
+
return onComplete(false, true);
}
@@ -637,6 +654,17 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
if (timeoutObj != null)
cctx.time().removeTimeoutObject(timeoutObj);
+ /** Ensures what waiters for ready {@link mappings} will be unblocked if error has occurred while mapping. */
+ if (tx != null) {
+ synchronized (this) {
+ if (!mappingsReady) {
+ mappingsReady = true;
+
+ notifyAll();
+ }
+ }
+ }
+
return true;
}
@@ -890,224 +918,234 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
boolean remap,
boolean topLocked
) throws IgniteCheckedException {
- AffinityTopologyVersion topVer = this.topVer;
+ try {
+ AffinityTopologyVersion topVer = this.topVer;
- assert topVer != null;
+ assert topVer != null;
- assert topVer.topologyVersion() > 0;
+ assert topVer.topologyVersion() > 0;
- if (CU.affinityNodes(cctx, topVer).isEmpty()) {
- onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
- "(all partition nodes left the grid): " + cctx.name()));
+ if (CU.affinityNodes(cctx, topVer).isEmpty()) {
+ onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
+ "(all partition nodes left the grid): " + cctx.name()));
- return;
- }
+ return;
+ }
- boolean clientNode = cctx.kernalContext().clientNode();
+ boolean clientNode = cctx.kernalContext().clientNode();
- assert !remap || (clientNode && (tx == null || !tx.hasRemoteLocks()));
+ assert !remap || (clientNode && (tx == null || !tx.hasRemoteLocks()));
- // First assume this node is primary for all keys passed in.
- if (!clientNode && mapAsPrimary(keys, topVer))
- return;
+ // First assume this node is primary for all keys passed in.
+ if (!clientNode && mapAsPrimary(keys, topVer))
+ return;
- mappings = new ArrayDeque<>();
+ mappings = new ArrayDeque<>();
- // Assign keys to primary nodes.
- GridNearLockMapping map = null;
+ // Assign keys to primary nodes.
+ GridNearLockMapping map = null;
- for (KeyCacheObject key : keys) {
- GridNearLockMapping updated = map(key, map, topVer);
+ for (KeyCacheObject key : keys) {
+ GridNearLockMapping updated = map(key, map, topVer);
- // If new mapping was created, add to collection.
- if (updated != map) {
- mappings.add(updated);
+ // If new mapping was created, add to collection.
+ if (updated != map) {
+ mappings.add(updated);
- if (tx != null && updated.node().isLocal())
- tx.colocatedLocallyMapped(true);
- }
+ if (tx != null && updated.node().isLocal())
+ tx.colocatedLocallyMapped(true);
+ }
- map = updated;
- }
+ map = updated;
+ }
- if (isDone()) {
- if (log.isDebugEnabled())
- log.debug("Abandoning (re)map because future is done: " + this);
+ if (isDone()) {
+ if (log.isDebugEnabled())
+ log.debug("Abandoning (re)map because future is done: " + this);
- return;
- }
+ return;
+ }
- if (log.isDebugEnabled())
- log.debug("Starting (re)map for mappings [mappings=" + mappings + ", fut=" + this + ']');
+ if (log.isDebugEnabled())
+ log.debug("Starting (re)map for mappings [mappings=" + mappings + ", fut=" + this + ']');
- boolean hasRmtNodes = false;
+ boolean hasRmtNodes = false;
- boolean first = true;
+ boolean first = true;
- // Create mini futures.
- for (Iterator<GridNearLockMapping> iter = mappings.iterator(); iter.hasNext(); ) {
- GridNearLockMapping mapping = iter.next();
+ // Create mini futures.
+ for (Iterator<GridNearLockMapping> iter = mappings.iterator(); iter.hasNext(); ) {
+ GridNearLockMapping mapping = iter.next();
- ClusterNode node = mapping.node();
- Collection<KeyCacheObject> mappedKeys = mapping.mappedKeys();
+ ClusterNode node = mapping.node();
+ Collection<KeyCacheObject> mappedKeys = mapping.mappedKeys();
- boolean loc = node.equals(cctx.localNode());
+ boolean loc = node.equals(cctx.localNode());
- assert !mappedKeys.isEmpty();
+ assert !mappedKeys.isEmpty();
- GridNearLockRequest req = null;
+ GridNearLockRequest req = null;
- Collection<KeyCacheObject> distributedKeys = new ArrayList<>(mappedKeys.size());
+ Collection<KeyCacheObject> distributedKeys = new ArrayList<>(mappedKeys.size());
- for (KeyCacheObject key : mappedKeys) {
- IgniteTxKey txKey = cctx.txKey(key);
+ for (KeyCacheObject key : mappedKeys) {
+ IgniteTxKey txKey = cctx.txKey(key);
- GridDistributedCacheEntry entry = null;
+ GridDistributedCacheEntry entry = null;
- if (tx != null) {
- IgniteTxEntry txEntry = tx.entry(txKey);
+ if (tx != null) {
+ IgniteTxEntry txEntry = tx.entry(txKey);
- if (txEntry != null) {
- entry = (GridDistributedCacheEntry)txEntry.cached();
+ if (txEntry != null) {
+ entry = (GridDistributedCacheEntry)txEntry.cached();
- if (entry != null && loc == entry.detached()) {
- entry = cctx.colocated().entryExx(key, topVer, true);
+ if (entry != null && loc == entry.detached()) {
+ entry = cctx.colocated().entryExx(key, topVer, true);
- txEntry.cached(entry);
+ txEntry.cached(entry);
+ }
}
}
- }
- boolean explicit;
+ boolean explicit;
- while (true) {
- try {
- if (entry == null)
- entry = cctx.colocated().entryExx(key, topVer, true);
+ while (true) {
+ try {
+ if (entry == null)
+ entry = cctx.colocated().entryExx(key, topVer, true);
- if (!cctx.isAll(entry, filter)) {
- if (log.isDebugEnabled())
- log.debug("Entry being locked did not pass filter (will not lock): " + entry);
+ if (!cctx.isAll(entry, filter)) {
+ if (log.isDebugEnabled())
+ log.debug("Entry being locked did not pass filter (will not lock): " + entry);
- onComplete(false, false);
+ onComplete(false, false);
- return;
- }
+ return;
+ }
- assert loc ^ entry.detached() : "Invalid entry [loc=" + loc + ", entry=" + entry + ']';
+ assert loc ^ entry.detached() : "Invalid entry [loc=" + loc + ", entry=" + entry + ']';
- GridCacheMvccCandidate cand = addEntry(entry);
+ GridCacheMvccCandidate cand = addEntry(entry);
- // Will either return value from dht cache or null if this is a miss.
- IgniteBiTuple<GridCacheVersion, CacheObject> val = entry.detached() ? null :
- ((GridDhtCacheEntry)entry).versionedValue(topVer);
+ // Will either return value from dht cache or null if this is a miss.
+ IgniteBiTuple<GridCacheVersion, CacheObject> val = entry.detached() ? null :
+ ((GridDhtCacheEntry)entry).versionedValue(topVer);
- GridCacheVersion dhtVer = null;
+ GridCacheVersion dhtVer = null;
- if (val != null) {
- dhtVer = val.get1();
+ if (val != null) {
+ dhtVer = val.get1();
- valMap.put(key, val);
- }
+ valMap.put(key, val);
+ }
- if (cand != null && !cand.reentry()) {
- if (req == null) {
- boolean clientFirst = false;
+ if (cand != null && !cand.reentry()) {
+ if (req == null) {
+ boolean clientFirst = false;
+
+ if (first) {
+ clientFirst = clientNode &&
+ !topLocked &&
+ (tx == null || !tx.hasRemoteLocks());
+
+ first = false;
+ }
+
+ assert !implicitTx() && !implicitSingleTx() : tx;
+
+ req = new GridNearLockRequest(
+ cctx.cacheId(),
+ topVer,
+ cctx.nodeId(),
+ threadId,
+ futId,
+ lockVer,
+ inTx(),
+ read,
+ retval,
+ isolation(),
+ isInvalidate(),
+ timeout,
+ mappedKeys.size(),
+ inTx() ? tx.size() : mappedKeys.size(),
+ inTx() && tx.syncMode() == FULL_SYNC,
+ inTx() ? tx.subjectId() : null,
+ inTx() ? tx.taskNameHash() : 0,
+ read ? createTtl : -1L,
+ read ? accessTtl : -1L,
+ skipStore,
+ keepBinary,
+ clientFirst,
+ false,
+ cctx.deploymentEnabled());
+
+ mapping.request(req);
+ }
- if (first) {
- clientFirst = clientNode &&
- !topLocked &&
- (tx == null || !tx.hasRemoteLocks());
+ distributedKeys.add(key);
- first = false;
- }
+ if (tx != null)
+ tx.addKeyMapping(txKey, mapping.node());
- assert !implicitTx() && !implicitSingleTx() : tx;
-
- req = new GridNearLockRequest(
- cctx.cacheId(),
- topVer,
- cctx.nodeId(),
- threadId,
- futId,
- lockVer,
- inTx(),
- read,
+ req.addKeyBytes(
+ key,
retval,
- isolation(),
- isInvalidate(),
- timeout,
- mappedKeys.size(),
- inTx() ? tx.size() : mappedKeys.size(),
- inTx() && tx.syncMode() == FULL_SYNC,
- inTx() ? tx.subjectId() : null,
- inTx() ? tx.taskNameHash() : 0,
- read ? createTtl : -1L,
- read ? accessTtl : -1L,
- skipStore,
- keepBinary,
- clientFirst,
- false,
- cctx.deploymentEnabled());
-
- mapping.request(req);
+ dhtVer, // Include DHT version to match remote DHT entry.
+ cctx);
}
- distributedKeys.add(key);
+ explicit = inTx() && cand == null;
- if (tx != null)
+ if (explicit)
tx.addKeyMapping(txKey, mapping.node());
- req.addKeyBytes(
- key,
- retval,
- dhtVer, // Include DHT version to match remote DHT entry.
- cctx);
+ break;
}
+ catch (GridCacheEntryRemovedException ignored) {
+ if (log.isDebugEnabled())
+ log.debug("Got removed entry in lockAsync(..) method (will retry): " + entry);
- explicit = inTx() && cand == null;
-
- if (explicit)
- tx.addKeyMapping(txKey, mapping.node());
-
- break;
+ entry = null;
+ }
}
- catch (GridCacheEntryRemovedException ignored) {
- if (log.isDebugEnabled())
- log.debug("Got removed entry in lockAsync(..) method (will retry): " + entry);
- entry = null;
+ // Mark mapping explicit lock flag.
+ if (explicit) {
+ boolean marked = tx != null && tx.markExplicit(node.id());
+
+ assert tx == null || marked;
}
}
- // Mark mapping explicit lock flag.
- if (explicit) {
- boolean marked = tx != null && tx.markExplicit(node.id());
+ if (!distributedKeys.isEmpty()) {
+ mapping.distributedKeys(distributedKeys);
- assert tx == null || marked;
+ hasRmtNodes |= !mapping.node().isLocal();
}
- }
-
- if (!distributedKeys.isEmpty()) {
- mapping.distributedKeys(distributedKeys);
+ else {
+ assert mapping.request() == null;
- hasRmtNodes |= !mapping.node().isLocal();
+ iter.remove();
+ }
}
- else {
- assert mapping.request() == null;
- iter.remove();
+ if (hasRmtNodes) {
+ trackable = true;
+
+ if (!remap && !cctx.mvcc().addFuture(this))
+ throw new IllegalStateException("Duplicate future ID: " + this);
}
+ else
+ trackable = false;
}
+ finally {
+ /** Notify ready {@link mappings} waiters. See {@link #cancel()} */
+ if (tx != null) {
+ mappingsReady = true;
- if (hasRmtNodes) {
- trackable = true;
-
- if (!remap && !cctx.mvcc().addFuture(this))
- throw new IllegalStateException("Duplicate future ID: " + this);
+ notifyAll();
+ }
}
- else
- trackable = false;
proceedMapping();
}
@@ -1137,11 +1175,27 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
throws IgniteCheckedException {
GridNearLockMapping map;
+ // Fail fast if future is completed (in case of async rollback)
+ if (isDone()) {
+ clear();
+
+ return;
+ }
+
+ // Fail fast if the transaction is timed out.
+ if (tx != null && tx.remainingTime() == -1) {
+ GridDhtColocatedLockFuture.this.onDone(false, tx.timeoutException());
+
+ clear();
+
+ return;
+ }
+
synchronized (this) {
map = mappings.poll();
}
- // If there are no more mappings to process, complete the future.
+ // If there are no more mappings to process or prepare has timed out, complete the future.
if (map == null)
return;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
index c16a934..ffde0f3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
@@ -38,7 +38,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartit
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxMapping;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
-import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.C1;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
index ede8a4e..2974563 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
@@ -412,19 +412,19 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
}
if (!commit && !clearThreadMap)
- tryRollbackAsync(onTimeout); // Asynchronous rollback.
+ rollbackAsyncSafe(onTimeout);
else
doFinish(commit, clearThreadMap);
}
/**
- * Does async rollback when it's safe.
- * If current future is not lock future (enlist future) waits until completion and tries again.
- * Else terminates or waits for lock future depending on rollback mode.
+ * Rollback tx when it's safe.
+ * If current future is not lock future (enlist future) wait until completion and tries again.
+ * Else cancel lock future (onTimeout=false) or wait for completion due to deadlock detection (onTimeout=true).
*
* @param onTimeout If {@code true} called from timeout handler.
*/
- private void tryRollbackAsync(boolean onTimeout) {
+ private void rollbackAsyncSafe(boolean onTimeout) {
IgniteInternalFuture<?> curFut = tx.tryRollbackAsync();
if (curFut == null) { // Safe to rollback.
@@ -447,7 +447,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
try {
fut.get();
- tryRollbackAsync(onTimeout);
+ rollbackAsyncSafe(onTimeout);
}
catch (IgniteCheckedException e) {
doFinish(false, false);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxInfo.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxInfo.java
index 03de5b0..be60a5e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxInfo.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxInfo.java
@@ -26,9 +26,12 @@ import java.time.format.DateTimeFormatter;
import java.util.Collection;
import java.util.TimeZone;
import java.util.UUID;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.visor.VisorDataTransferObject;
+import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
@@ -82,6 +85,9 @@ public class VisorTxInfo extends VisorDataTransferObject {
/** */
private Collection<UUID> masterNodeIds;
+ /** */
+ private AffinityTopologyVersion topVer;
+
/**
* Default constructor.
*/
@@ -103,7 +109,7 @@ public class VisorTxInfo extends VisorDataTransferObject {
*/
public VisorTxInfo(IgniteUuid xid, long startTime, long duration, TransactionIsolation isolation,
TransactionConcurrency concurrency, long timeout, String lb, Collection<UUID> primaryNodes,
- TransactionState state, int size, IgniteUuid nearXid, Collection<UUID> masterNodeIds) {
+ TransactionState state, int size, IgniteUuid nearXid, Collection<UUID> masterNodeIds, AffinityTopologyVersion topVer) {
this.xid = xid;
this.startTime = startTime;
this.duration = duration;
@@ -116,11 +122,12 @@ public class VisorTxInfo extends VisorDataTransferObject {
this.size = size;
this.nearXid = nearXid;
this.masterNodeIds = masterNodeIds;
+ this.topVer = topVer;
}
/** {@inheritDoc} */
@Override public byte getProtocolVersion() {
- return V2;
+ return V3;
}
/** */
@@ -154,6 +161,11 @@ public class VisorTxInfo extends VisorDataTransferObject {
}
/** */
+ public AffinityTopologyVersion getTopologyVersion() {
+ return topVer;
+ }
+
+ /** */
public long getTimeout() {
return timeout;
}
@@ -202,6 +214,8 @@ public class VisorTxInfo extends VisorDataTransferObject {
U.writeGridUuid(out, nearXid);
U.writeCollection(out, masterNodeIds);
out.writeLong(startTime);
+ out.writeLong(topVer == null ? -1 : topVer.topologyVersion());
+ out.writeInt(topVer == null ? -1 : topVer.minorTopologyVersion());
}
/** {@inheritDoc} */
@@ -217,12 +231,48 @@ public class VisorTxInfo extends VisorDataTransferObject {
size = in.readInt();
if (protoVer >= V2) {
nearXid = U.readGridUuid(in);
-
masterNodeIds = U.readCollection(in);
-
startTime = in.readLong();
}
+ if (protoVer >= V3) {
+ long topVer = in.readLong();
+ int minorTopVer = in.readInt();
+
+ if (topVer != -1)
+ this.topVer = new AffinityTopologyVersion(topVer, minorTopVer);
+ }
+ }
+ /**
+ * Get tx info as user string.
+ *
+ * @return User string.
+ */
+ public String toUserString() {
+ return " Tx: [xid=" + getXid() +
+ ", label=" + getLabel() +
+ ", state=" + getState() +
+ ", startTime=" + getFormattedStartTime() +
+ ", duration=" + getDuration() / 1000 +
+ ", isolation=" + getIsolation() +
+ ", concurrency=" + getConcurrency() +
+ ", topVer=" + (getTopologyVersion() == null ? "N/A" : getTopologyVersion()) +
+ ", timeout=" + getTimeout() +
+ ", size=" + getSize() +
+ ", dhtNodes=" + (getPrimaryNodes() == null ? "N/A" :
+ F.transform(getPrimaryNodes(), new IgniteClosure<UUID, String>() {
+ @Override public String apply(UUID id) {
+ return U.id8(id);
+ }
+ })) +
+ ", nearXid=" + getNearXid() +
+ ", parentNodeIds=" + (getMasterNodeIds() == null ? "N/A" :
+ F.transform(getMasterNodeIds(), new IgniteClosure<UUID, String>() {
+ @Override public String apply(UUID id) {
+ return U.id8(id);
+ }
+ })) +
+ ']';
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxTask.java
index 9919b7d..23d1663 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxTask.java
@@ -183,7 +183,7 @@ public class VisorTxTask extends VisorMultiNodeTask<VisorTxTaskArg, Map<ClusterN
private static final TxKillClosure NEAR_KILL_CLOSURE = new NearKillClosure();
/** */
- private static final TxKillClosure LOCAL_KILL_CLOSURE = NEAR_KILL_CLOSURE;
+ private static final TxKillClosure LOCAL_KILL_CLOSURE = new LocalKillClosure();
/** */
private static final TxKillClosure REMOTE_KILL_CLOSURE = new RemoteKillClosure();
@@ -312,7 +312,7 @@ public class VisorTxTask extends VisorMultiNodeTask<VisorTxTaskArg, Map<ClusterN
infos.add(new VisorTxInfo(locTx.xid(), locTx.startTime(), duration, locTx.isolation(), locTx.concurrency(),
locTx.timeout(), lb, mappings, locTx.state(),
- size, locTx.nearXidVersion().asGridUuid(), locTx.masterNodeIds()));
+ size, locTx.nearXidVersion().asGridUuid(), locTx.masterNodeIds(), locTx.topologyVersionSnapshot()));
if (arg.getOperation() == VisorTxOperation.KILL)
killClo.apply(locTx, tm);
@@ -394,7 +394,7 @@ public class VisorTxTask extends VisorMultiNodeTask<VisorTxTaskArg, Map<ClusterN
IgniteBiClosure<IgniteInternalTx, IgniteTxManager, IgniteInternalFuture<IgniteInternalTx>> {
}
- /** Kills near or local tx. */
+ /** Kills near tx. */
private static class NearKillClosure implements TxKillClosure {
/** */
private static final long serialVersionUID = 0L;
@@ -402,7 +402,19 @@ public class VisorTxTask extends VisorMultiNodeTask<VisorTxTaskArg, Map<ClusterN
/** {@inheritDoc} */
@Override public IgniteInternalFuture<IgniteInternalTx> apply(IgniteInternalTx tx, IgniteTxManager tm) {
return tx.isRollbackOnly() || tx.state() == COMMITTING || tx.state() == COMMITTED ?
- new GridFinishedFuture<>() : tx.rollbackAsync();
+ new GridFinishedFuture<>() : ((GridNearTxLocal)tx).rollbackNearTxLocalAsync(false, false);
+ }
+ }
+
+ /** Kills local tx. */
+ private static class LocalKillClosure implements TxKillClosure {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<IgniteInternalTx> apply(IgniteInternalTx tx, IgniteTxManager tm) {
+ return tx.isRollbackOnly() || tx.state() == COMMITTING || tx.state() == COMMITTED ?
+ new GridFinishedFuture<>() : ((GridDhtTxLocal)tx).rollbackDhtLocalAsync();
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackAsyncTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackAsyncTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackAsyncTest.java
index 4626dcd..646c3f0 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackAsyncTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackAsyncTest.java
@@ -42,24 +42,34 @@ import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.GridCacheFuture;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFinishRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
+import org.apache.ignite.internal.util.lang.GridAbsPredicate;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.internal.visor.VisorTaskArgument;
+import org.apache.ignite.internal.visor.tx.VisorTxInfo;
+import org.apache.ignite.internal.visor.tx.VisorTxOperation;
+import org.apache.ignite.internal.visor.tx.VisorTxTask;
+import org.apache.ignite.internal.visor.tx.VisorTxTaskArg;
+import org.apache.ignite.internal.visor.tx.VisorTxTaskResult;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
@@ -75,6 +85,7 @@ import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.configuration.WALMode.LOG_ONLY;
import static org.apache.ignite.testframework.GridTestUtils.runAsync;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
import static org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
@@ -864,6 +875,122 @@ public class TxRollbackAsyncTest extends GridCommonAbstractTest {
}
/**
+ *
+ */
+ public void testRollbackOnTopologyLockPessimistic() throws Exception {
+ final Ignite client = startClient();
+
+ Ignite crd = grid(0);
+
+ List<Integer> keys = primaryKeys(grid(1).cache(CACHE_NAME), 1);
+
+ assertTrue(crd.cluster().localNode().order() == 1);
+
+ CountDownLatch txLatch = new CountDownLatch(1);
+ CountDownLatch tx2Latch = new CountDownLatch(1);
+ CountDownLatch commitLatch = new CountDownLatch(1);
+
+ // Start tx holding topology.
+ IgniteInternalFuture txFut = runAsync(new Runnable() {
+ @Override public void run() {
+ List<Integer> keys = primaryKeys(grid(0).cache(CACHE_NAME), 1);
+
+ try (Transaction tx = client.transactions().txStart()) {
+ client.cache(CACHE_NAME).put(keys.get(0), 0);
+
+ txLatch.countDown();
+
+ U.awaitQuiet(commitLatch);
+
+ tx.commit();
+
+ fail();
+ }
+ catch (Exception e) {
+ // Expected.
+ }
+ }
+ });
+
+ U.awaitQuiet(txLatch);
+
+ crd.events().localListen(new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event evt) {
+ runAsync(new Runnable() {
+ @Override public void run() {
+ try(Transaction tx = crd.transactions().withLabel("testLbl").txStart()) {
+ // Wait for node start.
+ waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return crd.cluster().topologyVersion() != GRID_CNT +
+ /** client node */ 1 + /** stop server node */ 1 + /** start server node */ 1;
+ }
+ }, 10_000);
+
+ tx2Latch.countDown();
+
+ crd.cache(CACHE_NAME).put(keys.get(0), 0);
+
+ tx.commit();
+
+ fail();
+ }
+ catch (Exception e) {
+ // Expected.
+ }
+ }
+ });
+
+ return false;
+ }
+ }, EventType.EVT_NODE_FAILED, EventType.EVT_NODE_LEFT);
+
+ IgniteInternalFuture restartFut = runAsync(new Runnable() {
+ @Override public void run() {
+ stopGrid(2);
+
+ try {
+ startGrid(2);
+ }
+ catch (Exception e) {
+ fail();
+ }
+ }
+ });
+
+ U.awaitQuiet(tx2Latch);
+
+ // Rollback tx using kill task.
+ VisorTxTaskArg arg =
+ new VisorTxTaskArg(VisorTxOperation.KILL, null, null, null, null, null, null, null, null, null);
+
+ Map<ClusterNode, VisorTxTaskResult> res = client.compute(client.cluster().forPredicate(F.alwaysTrue())).
+ execute(new VisorTxTask(), new VisorTaskArgument<>(client.cluster().localNode().id(), arg, false));
+
+ int expCnt = 0;
+
+ for (Map.Entry<ClusterNode, VisorTxTaskResult> entry : res.entrySet()) {
+ if (entry.getValue().getInfos().isEmpty())
+ continue;
+
+ for (VisorTxInfo info : entry.getValue().getInfos()) {
+ log.info(info.toUserString());
+
+ expCnt++;
+ }
+ }
+
+ assertEquals("Expecting 2 transactions", 2, expCnt);
+
+ commitLatch.countDown();
+
+ txFut.get();
+ restartFut.get();
+
+ checkFutures();
+ }
+
+ /**
* Locks entry in tx and delays commit until signalled.
*
* @param node Near node.
@@ -896,22 +1023,6 @@ public class TxRollbackAsyncTest extends GridCommonAbstractTest {
}
/**
- * Checks if all tx futures are finished.
- */
- private void checkFutures() {
- for (Ignite ignite : G.allGrids()) {
- IgniteEx ig = (IgniteEx)ignite;
-
- final Collection<GridCacheFuture<?>> futs = ig.context().cache().context().mvcc().activeFutures();
-
- for (GridCacheFuture<?> fut : futs)
- log.info("Waiting for future: " + fut);
-
- assertTrue("Expecting no active futures: node=" + ig.localNode().id(), futs.isEmpty());
- }
- }
-
- /**
* @param tx Tx to rollback.
*/
private IgniteInternalFuture<?> rollbackAsync(final Transaction tx) throws Exception {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
index 9c37cfa..d6f3ae5 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.internal.processors.cache.transactions;
import java.util.Collection;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -35,6 +37,8 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
@@ -42,12 +46,23 @@ import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.internal.visor.VisorTaskArgument;
+import org.apache.ignite.internal.visor.tx.VisorTxInfo;
+import org.apache.ignite.internal.visor.tx.VisorTxOperation;
+import org.apache.ignite.internal.visor.tx.VisorTxTask;
+import org.apache.ignite.internal.visor.tx.VisorTxTaskArg;
+import org.apache.ignite.internal.visor.tx.VisorTxTaskResult;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -62,6 +77,7 @@ import org.apache.ignite.transactions.TransactionTimeoutException;
import static java.lang.Thread.sleep;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.testframework.GridTestUtils.runAsync;
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
@@ -88,6 +104,8 @@ public class TxRollbackOnTimeoutTest extends GridCommonAbstractTest {
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+ cfg.setConsistentId(igniteInstanceName);
+
((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(IP_FINDER);
cfg.setCommunicationSpi(new TestRecordingCommunicationSpi());
@@ -298,7 +316,7 @@ public class TxRollbackOnTimeoutTest extends GridCommonAbstractTest {
final CountDownLatch l = new CountDownLatch(2);
- IgniteInternalFuture<?> fut1 = GridTestUtils.runAsync(new Runnable() {
+ IgniteInternalFuture<?> fut1 = runAsync(new Runnable() {
@Override public void run() {
try {
try (Transaction tx = node1.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 5000, 2)) {
@@ -322,7 +340,7 @@ public class TxRollbackOnTimeoutTest extends GridCommonAbstractTest {
}
}, "First");
- IgniteInternalFuture<?> fut2 = GridTestUtils.runAsync(new Runnable() {
+ IgniteInternalFuture<?> fut2 = runAsync(new Runnable() {
@Override public void run() {
try (Transaction tx = node2.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 0, 2)) {
node2.cache(CACHE_NAME).put(2, 2);
@@ -812,7 +830,7 @@ public class TxRollbackOnTimeoutTest extends GridCommonAbstractTest {
final int recordsCnt = 5;
- IgniteInternalFuture<?> fut1 = GridTestUtils.runAsync(new Runnable() {
+ IgniteInternalFuture<?> fut1 = runAsync(new Runnable() {
@Override public void run() {
try (Transaction tx = near.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, timeout, 0)) {
try {
@@ -860,7 +878,7 @@ public class TxRollbackOnTimeoutTest extends GridCommonAbstractTest {
}
}, "First");
- IgniteInternalFuture<?> fut2 = GridTestUtils.runAsync(new Runnable() {
+ IgniteInternalFuture<?> fut2 = runAsync(new Runnable() {
@Override public void run() {
U.awaitQuiet(blocked);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTopologyChangeTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTopologyChangeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTopologyChangeTest.java
index 2542cdb..13c5e41 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTopologyChangeTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTopologyChangeTest.java
@@ -209,20 +209,4 @@ public class TxRollbackOnTopologyChangeTest extends GridCommonAbstractTest {
checkFutures();
}
-
- /**
- * Checks if all tx futures are finished.
- */
- private void checkFutures() {
- for (Ignite ignite : G.allGrids()) {
- IgniteEx ig = (IgniteEx)ignite;
-
- final Collection<GridCacheFuture<?>> futs = ig.context().cache().context().mvcc().activeFutures();
-
- for (GridCacheFuture<?> fut : futs)
- log.info("Waiting for future: " + fut);
-
- assertTrue("Expecting no active futures: node=" + ig.localNode().id(), futs.isEmpty());
- }
- }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
index 8591f0c..55086f3 100755
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
@@ -1995,4 +1995,27 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
new VisorTaskArgument<>(node.id(), taskArg, false)
);
}
+
+ /**
+ * Checks if all txs and mvcc futures are finished.
+ */
+ protected void checkFutures() {
+ for (Ignite ignite : G.allGrids()) {
+ IgniteEx ig = (IgniteEx)ignite;
+
+ final Collection<GridCacheFuture<?>> futs = ig.context().cache().context().mvcc().activeFutures();
+
+ for (GridCacheFuture<?> fut : futs)
+ log.info("Waiting for future: " + fut);
+
+ assertTrue("Expecting no active futures: node=" + ig.localNode().id(), futs.isEmpty());
+
+ Collection<IgniteInternalTx> txs = ig.context().cache().context().tm().activeTransactions();
+
+ for (IgniteInternalTx tx : txs)
+ log.info("Waiting for tx: " + tx);
+
+ assertTrue("Expecting no active transactions: node=" + ig.localNode().id(), txs.isEmpty());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3882ec04/modules/core/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java b/modules/core/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
index f612a96..8af7a10 100644
--- a/modules/core/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
@@ -1415,29 +1415,6 @@ public class GridCommandHandlerTest extends GridCommonAbstractTest {
}, 4, "tx-thread");
}
- /**
- * Checks if all tx futures are finished.
- */
- private void checkFutures() {
- for (Ignite ignite : G.allGrids()) {
- IgniteEx ig = (IgniteEx)ignite;
-
- final Collection<GridCacheFuture<?>> futs = ig.context().cache().context().mvcc().activeFutures();
-
- for (GridCacheFuture<?> fut : futs)
- log.info("Waiting for future: " + fut);
-
- assertTrue("Expecting no active futures: node=" + ig.localNode().id(), futs.isEmpty());
-
- Collection<IgniteInternalTx> txs = ig.context().cache().context().tm().activeTransactions();
-
- for (IgniteInternalTx tx : txs)
- log.info("Waiting for tx: " + tx);
-
- assertTrue("Expecting no active transactions: node=" + ig.localNode().id(), txs.isEmpty());
- }
- }
-
/** */
private static class IncrementClosure implements EntryProcessor<Long, Long, Void> {
/** {@inheritDoc} */