You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ir...@apache.org on 2019/09/02 16:56:14 UTC
[ignite] branch master updated: IGNITE-12038 Fix several failing
tests after IGNITE-10078 - Fixes #6746.
This is an automated email from the ASF dual-hosted git repository.
irakov 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 da4f508 IGNITE-12038 Fix several failing tests after IGNITE-10078 - Fixes #6746.
da4f508 is described below
commit da4f5087c8505f46319261027a10eac1f72c1b0f
Author: Alexey Scherbakov <al...@gmail.com>
AuthorDate: Mon Sep 2 19:55:54 2019 +0300
IGNITE-12038 Fix several failing tests after IGNITE-10078 - Fixes #6746.
Signed-off-by: Ivan Rakov <ir...@apache.org>
---
.../affinity/AffinityTopologyVersion.java | 18 +-
.../processors/cache/GridCacheIoManager.java | 3 +-
.../processors/cache/GridCacheMapEntry.java | 2 +
.../cache/GridCacheSharedTtlCleanupManager.java | 6 +-
.../internal/processors/cache/GridCacheUtils.java | 1 +
.../cache/PartitionTxUpdateCounterImpl.java | 48 +-
.../GridDistributedTxRemoteAdapter.java | 41 ++
.../dht/GridDhtTransactionalCacheAdapter.java | 19 +-
.../distributed/dht/GridDhtTxPrepareFuture.java | 12 +-
.../dht/colocated/GridDhtColocatedCache.java | 4 +-
.../dht/colocated/GridDhtColocatedLockFuture.java | 163 +++---
.../dht/preloader/GridDhtPartitionDemander.java | 14 +-
.../preloader/GridDhtPartitionsExchangeFuture.java | 34 +-
.../dht/preloader/GridDhtPreloader.java | 7 +-
.../dht/topology/GridDhtLocalPartition.java | 81 ++-
.../dht/topology/GridDhtPartitionTopologyImpl.java | 39 +-
.../dht/topology/PartitionsEvictManager.java | 6 +-
.../cache/distributed/near/GridNearLockFuture.java | 232 ++++----
.../distributed/near/GridNearLockRequest.java | 7 +
.../distributed/near/GridNearLockResponse.java | 53 +-
.../near/GridNearOptimisticTxPrepareFuture.java | 27 +-
.../GridNearOptimisticTxPrepareFutureAdapter.java | 2 +-
.../cache/distributed/near/GridNearTxLocal.java | 21 +-
.../msg/PartitionCountersNeighborcastRequest.java | 32 +-
.../msg/PartitionCountersNeighborcastResponse.java | 31 +-
.../cache/persistence/GridCacheOffheapManager.java | 11 +-
.../cache/transactions/IgniteTxHandler.java | 2 +-
.../cache/transactions/IgniteTxLocalAdapter.java | 30 +-
.../PartitionCountersNeighborcastFuture.java | 2 +-
.../ignite/cache/ResetLostPartitionTest.java | 23 +-
.../CacheDataLossOnPartitionMoveTest.java | 2 +-
.../CachePartitionLossDetectionOnNodeLeftTest.java | 3 -
.../CachePartitionLostWhileClearingTest.java | 172 ++++++
.../GridCacheRebalancingSyncSelfTest.java | 5 +
...ocalWalModeChangeDuringRebalancingSelfTest.java | 22 +-
...IgnitePdsCacheWalDisabledOnRebalancingTest.java | 66 ++-
.../db/file/IgnitePdsDiskErrorsRecoveringTest.java | 2 +
.../TxCrossCacheMapOnInvalidTopologyTest.java | 316 +++++++++++
.../TxPartitionCounterStateAbstractTest.java | 21 +-
...ounterStateConsistencyHistoryRebalanceTest.java | 162 ------
.../TxPartitionCounterStateConsistencyTest.java | 615 +++++++++++++++++++--
.../TxRollbackOnMapOnInvalidTopologyTest.java | 189 +++++++
.../spi/discovery/tcp/BlockTcpDiscoverySpi.java | 100 ++++
.../TcpDiscoveryReconnectUnstableTopologyTest.java | 78 ---
.../junits/common/GridCommonAbstractTest.java | 44 +-
.../testsuites/IgniteCacheMvccTestSuite7.java | 4 +
.../ignite/testsuites/IgniteCacheTestSuite6.java | 3 +
.../ignite/testsuites/IgniteCacheTestSuite7.java | 6 +
...aryMetadataConcurrentUpdateWithIndexesTest.java | 80 +--
.../cache/mvcc/CacheMvccTxRecoveryTest.java | 78 +++
50 files changed, 2189 insertions(+), 750 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/AffinityTopologyVersion.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/AffinityTopologyVersion.java
index 2c02f26..3b9119b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/AffinityTopologyVersion.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/AffinityTopologyVersion.java
@@ -117,10 +117,26 @@ public class AffinityTopologyVersion implements Comparable<AffinityTopologyVersi
* @param upper Upper bound.
* @return {@code True} if this topology version is within provided bounds (inclusive).
*/
- public boolean isBetween(AffinityTopologyVersion lower, AffinityTopologyVersion upper) {
+ public final boolean isBetween(AffinityTopologyVersion lower, AffinityTopologyVersion upper) {
return compareTo(lower) >= 0 && compareTo(upper) <= 0;
}
+ /**
+ * @param topVer Test version.
+ * @return {@code True} if this topology happens strictly after than {@code topVer}.
+ */
+ public final boolean after(AffinityTopologyVersion topVer) {
+ return compareTo(topVer) > 0;
+ }
+
+ /**
+ * @param topVer Test version.
+ * @return {@code True} if this topology happens strictly before than {@code topVer}.
+ */
+ public final boolean before(AffinityTopologyVersion topVer) {
+ return compareTo(topVer) < 0;
+ }
+
/** {@inheritDoc} */
@Override public void onAckReceived() {
// No-op.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
index 539e1d3..7170ac3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
@@ -236,7 +236,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
else {
AffinityTopologyVersion locAffVer = cctx.exchange().readyAffinityVersion();
- if (locAffVer.compareTo(lastAffChangedVer) < 0) {
+ if (locAffVer.before(lastAffChangedVer)) {
IgniteLogger log = cacheMsg.messageLogger(cctx);
if (log.isDebugEnabled()) {
@@ -889,6 +889,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
0,
req.classError(),
null,
+ false,
false);
sendResponseOnFailedMessage(nodeId, res, cctx, plc);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index 725c01a..f9388d2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -119,6 +119,7 @@ import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRA
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.UPDATE;
import static org.apache.ignite.internal.processors.cache.GridCacheUpdateAtomicResult.UpdateOutcome.INVOKE_NO_OP;
import static org.apache.ignite.internal.processors.cache.GridCacheUpdateAtomicResult.UpdateOutcome.REMOVE_NO_VAL;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.RENTING;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_MAX_SNAPSHOT;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.compareIgnoreOpCounter;
import static org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter.RowData.NO_KEY;
@@ -4280,6 +4281,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
@Nullable CacheDataRow row
) throws IgniteCheckedException {
assert lock.isHeldByCurrentThread();
+ assert localPartition() == null || localPartition().state() != RENTING : localPartition();
UpdateClosure closure = new UpdateClosure(this, val, ver, expireTime, predicate, row);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedTtlCleanupManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedTtlCleanupManager.java
index 3623fd9..4094559 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedTtlCleanupManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedTtlCleanupManager.java
@@ -171,8 +171,12 @@ public class GridCacheSharedTtlCleanupManager extends GridCacheSharedManagerAdap
return;
}
- if (!(t instanceof IgniteInterruptedCheckedException || t instanceof InterruptedException))
+ if (!(t instanceof IgniteInterruptedCheckedException || t instanceof InterruptedException)) {
+ if (isCancelled)
+ return;
+
err = t;
+ }
throw t;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
index a249914..d9ff104 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
@@ -865,6 +865,7 @@ public class GridCacheUtils {
", rollbackOnly=" + tx.isRollbackOnly() +
", nodeId=" + tx.nodeId() +
", timeout=" + tx.timeout() +
+ ", startTime=" + tx.startTime() +
", duration=" + (U.currentTimeMillis() - tx.startTime()) +
(tx instanceof GridNearTxLocal ? ", label=" + tx.label() : "") +
']';
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionTxUpdateCounterImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionTxUpdateCounterImpl.java
index 00d1a17..14eb004 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionTxUpdateCounterImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionTxUpdateCounterImpl.java
@@ -72,7 +72,10 @@ public class PartitionTxUpdateCounterImpl implements PartitionUpdateCounter {
private final AtomicLong cntr = new AtomicLong();
/** HWM. */
- private final AtomicLong reserveCntr = new AtomicLong();
+ protected final AtomicLong reserveCntr = new AtomicLong();
+
+ /** */
+ private boolean first = true;
/**
* Initial counter points to last sequential update after WAL recovery.
@@ -84,7 +87,7 @@ public class PartitionTxUpdateCounterImpl implements PartitionUpdateCounter {
@Override public void init(long initUpdCntr, @Nullable byte[] cntrUpdData) {
cntr.set(initUpdCntr);
- initCntr = initUpdCntr;
+ reserveCntr.set(initCntr = initUpdCntr);
queue = fromBytes(cntrUpdData);
}
@@ -117,24 +120,36 @@ public class PartitionTxUpdateCounterImpl implements PartitionUpdateCounter {
/** {@inheritDoc} */
@Override public synchronized void update(long val) throws IgniteCheckedException {
- // Absolute counter should be not less than last applied update.
- // Otherwise supplier doesn't contain some updates and rebalancing couldn't restore consistency.
- // Best behavior is to stop node by failure handler in such a case.
- if (!gaps().isEmpty() && val < highestAppliedCounter())
- throw new IgniteCheckedException("Failed to update the counter [newVal=" + val + ", curState=" + this + ']');
+ // Reserved update counter is updated only on exchange.
+ long cur = get();
- long cur = cntr.get();
+ // Always set reserved counter equal to max known counter.
+ long max = Math.max(val, cur);
- // Reserved update counter is updated only on exchange or in non-tx mode.
- reserveCntr.set(Math.max(cur, val));
+ if (reserveCntr.get() < max)
+ reserveCntr.set(max);
- if (val <= cur)
+ // Outdated counter (txs are possible before current topology future is finished if primary is not changed).
+ if (val < cur)
return;
+ // Absolute counter should be not less than last applied update.
+ // Otherwise supplier doesn't contain some updates and rebalancing couldn't restore consistency.
+ // Best behavior is to stop node by failure handler in such a case.
+ if (val < highestAppliedCounter())
+ throw new IgniteCheckedException("Failed to update the counter [newVal=" + val + ", curState=" + this + ']');
+
cntr.set(val);
- if (!queue.isEmpty())
- queue.clear();
+ /** If some holes are present at this point, thar means some update were missed on recovery and will be restored
+ * during rebalance. All gaps are safe to "forget".
+ * Should only do it for first PME (later missed updates on node left are reset in {@link #finalizeUpdateCounters}. */
+ if (first) {
+ if (!queue.isEmpty())
+ queue.clear();
+
+ first = false;
+ }
}
/** {@inheritDoc} */
@@ -214,6 +229,9 @@ public class PartitionTxUpdateCounterImpl implements PartitionUpdateCounter {
update(start, delta);
initCntr = get();
+
+ if (reserveCntr.get() < initCntr)
+ reserveCntr.set(initCntr);
}
/** */
@@ -258,11 +276,13 @@ public class PartitionTxUpdateCounterImpl implements PartitionUpdateCounter {
item = poll();
}
+ reserveCntr.set(get());
+
return gaps;
}
/** {@inheritDoc} */
- @Override public long reserve(long delta) {
+ @Override public synchronized long reserve(long delta) {
long cntr = get();
long reserved = reserveCntr.getAndAdd(delta);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index 22e3dd5..6936e816 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -21,6 +21,7 @@ import java.io.Externalizable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -50,6 +51,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheReturnCompletableWra
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.GridCacheUpdateTxResult;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxAdapter;
@@ -70,6 +72,7 @@ import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
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.LT;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.transactions.TransactionConcurrency;
@@ -83,6 +86,8 @@ import static org.apache.ignite.internal.processors.cache.GridCacheOperation.NOO
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.READ;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.RELOAD;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.UPDATE;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.RENTING;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.EVICTED;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_BACKUP;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_NONE;
import static org.apache.ignite.transactions.TransactionState.COMMITTED;
@@ -499,6 +504,9 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
cctx.database().checkpointReadLock();
+ // Reserved partitions (necessary to prevent race due to updates in RENTING state).
+ Set<GridDhtLocalPartition> reservedParts = new HashSet<>();
+
try {
assert !txState.mvccEnabled() || mvccSnapshot != null : "Mvcc is not initialized: " + this;
@@ -513,6 +521,36 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
for (IgniteTxEntry txEntry : entries) {
GridCacheContext cacheCtx = txEntry.context();
+ // Prevent stale updates.
+ GridDhtLocalPartition locPart =
+ cacheCtx.group().topology().localPartition(txEntry.cached().partition());
+
+ boolean reserved = false;
+
+ if (!near() && locPart != null && !reservedParts.contains(locPart) &&
+ (!(reserved = locPart.reserve()) || locPart.state() == RENTING)) {
+ LT.warn(log(), "Skipping update to partition that is concurrently evicting " +
+ "[grp=" + cacheCtx.group().cacheOrGroupName() + ", part=" + locPart + "]");
+
+ // Reserved RENTING partition.
+ if (reserved) {
+ assert locPart.state() != EVICTED && locPart.reservations() > 0;
+
+ reservedParts.add(locPart);
+ }
+
+ continue;
+ }
+
+ if (reserved) {
+ assert locPart.state() != EVICTED && locPart.reservations() > 0;
+
+ reservedParts.add(locPart);
+ }
+
+ assert near() || locPart == null ||
+ !(locPart.state() == RENTING || locPart.state() == EVICTED) : locPart;
+
boolean replicate = cacheCtx.isDrEnabled();
while (true) {
@@ -804,6 +842,9 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
throw err;
}
finally {
+ for (GridDhtLocalPartition locPart : reservedParts)
+ locPart.release();
+
cctx.database().checkpointReadUnlock();
if (wrapper != null)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index 3fa51f0..8fdc6fd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
@@ -1102,7 +1102,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
if (tx == null) {
tx = new GridDhtTxLocal(
ctx.shared(),
- req.topologyVersion(),
+ topology().readyTopologyVersion(),
nearNode.id(),
req.version(),
req.futureId(),
@@ -1351,7 +1351,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
0,
null,
topVer,
- ctx.deploymentEnabled());
+ ctx.deploymentEnabled(),
+ false);
try {
ctx.io().send(nearNode, res, ctx.ioPolicy());
@@ -1388,6 +1389,12 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
assert tx == null || tx.xidVersion().equals(mappedVer);
try {
+ // All subsequent lock requests must use actual topology version to avoid mapping on invalid primaries.
+ AffinityTopologyVersion clienRemapVer = req.firstClientRequest() &&
+ tx != null &&
+ topology().readyTopologyVersion().after(req.topologyVersion()) ?
+ topology().readyTopologyVersion() : null;
+
// Send reply back to originating near node.
GridNearLockResponse res = new GridNearLockResponse(ctx.cacheId(),
req.version(),
@@ -1396,8 +1403,9 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
tx != null && tx.onePhaseCommit(),
entries.size(),
err,
- null,
- ctx.deploymentEnabled());
+ clienRemapVer,
+ ctx.deploymentEnabled(),
+ clienRemapVer != null);
if (err == null) {
res.pending(localDhtPendingVersions(entries, mappedVer));
@@ -1508,7 +1516,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
entries.size(),
e,
null,
- ctx.deploymentEnabled());
+ ctx.deploymentEnabled(),
+ false);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
index ec8a451..6a42c1b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -76,6 +76,7 @@ import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.dr.GridDrType;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObjectAdapter;
import org.apache.ignite.internal.transactions.IgniteTxOptimisticCheckedException;
+import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
import org.apache.ignite.internal.util.GridLeanSet;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -704,7 +705,12 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
}
if (forceKeysFut == null || (forceKeysFut.isDone() && forceKeysFut.error() == null))
- prepare0();
+ try {
+ prepare0();
+ }
+ catch (IgniteTxRollbackCheckedException e) {
+ onError(e);
+ }
else {
forceKeysFut.listen(new CI1<IgniteInternalFuture<?>>() {
@Override public void apply(IgniteInternalFuture<?> f) {
@@ -1258,7 +1264,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
/**
*
*/
- private void prepare0() {
+ private void prepare0() throws IgniteTxRollbackCheckedException {
boolean error = false;
try {
@@ -1635,7 +1641,7 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
List<ClusterNode> dhtNodes = dht.topology().nodes(cached.partition(), tx.topologyVersion());
assert !dhtNodes.isEmpty() && dhtNodes.get(0).id().equals(cctx.localNodeId()) :
- "localNode = " + cctx.localNodeId() + ", dhtNodes = " + dhtNodes;
+ "cacheId=" + cacheCtx.cacheId() + ", localNode = " + cctx.localNodeId() + ", dhtNodes = " + dhtNodes;
if (log.isDebugEnabled())
log.debug("Mapping entry to DHT nodes [nodes=" + U.toShortString(dhtNodes) +
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
index a8a27d1..63a0582 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
@@ -136,7 +136,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
ctx.io().addCacheHandler(ctx.cacheId(), GridNearLockResponse.class, new CI2<UUID, GridNearLockResponse>() {
@Override public void apply(UUID nodeId, GridNearLockResponse res) {
- processLockResponse(nodeId, res);
+ processNearLockResponse(nodeId, res);
}
});
}
@@ -1201,7 +1201,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
* @param nodeId Node ID.
* @param res Response.
*/
- private void processLockResponse(UUID nodeId, GridNearLockResponse res) {
+ private void processNearLockResponse(UUID nodeId, GridNearLockResponse res) {
if (txLockMsgLog.isDebugEnabled())
txLockMsgLog.debug("Received near lock response [txId=" + res.version() + ", node=" + nodeId + ']');
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 9c5374a..6731405 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
@@ -829,7 +829,7 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
final GridDhtTopologyFuture fut;
- final boolean finish;
+ final boolean finished;
try {
if (cctx.topology().stopping()) {
@@ -843,9 +843,9 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
fut = cctx.topologyVersionFuture();
- finish = fut.isDone();
+ finished = fut.isDone();
- if (finish) {
+ if (finished) {
Throwable err = fut.validateCache(cctx, recovery, read, null, keys);
if (err != null) {
@@ -858,7 +858,7 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
if (remap) {
if (tx != null)
- tx.onRemap(topVer);
+ tx.onRemap(topVer, true);
synchronized (this) {
this.topVer = topVer;
@@ -882,7 +882,7 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
cctx.topology().readUnlock();
}
- if (finish) {
+ if (finished) {
map(keys, remap, false);
if (c != null)
@@ -1665,94 +1665,117 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
if (res.clientRemapVersion() != null) {
assert cctx.kernalContext().clientNode();
- IgniteInternalFuture<?> affFut =
- cctx.shared().exchange().affinityReadyFuture(res.clientRemapVersion());
+ if (res.compatibleRemapVersion()) {
+ if (tx != null) {
+ // Versions are compatible.
+ cctx.shared().exchange().
+ lastAffinityChangedTopologyVersion(res.clientRemapVersion(), tx.topologyVersionSnapshot());
- cctx.time().waitAsync(affFut, tx == null ? 0 : tx.remainingTime(), (e, timedOut) -> {
- if (errorOrTimeoutOnTopologyVersion(e, timedOut))
- return;
+ tx.onRemap(res.clientRemapVersion(), false);
- try {
- remap();
- }
- finally {
- cctx.shared().txContextReset();
- }
- });
- }
- else {
- int i = 0;
+ // Use remapped version for all subsequent mappings.
+ synchronized (GridDhtColocatedLockFuture.this) {
+ for (GridNearLockMapping mapping : mappings) {
+ GridNearLockRequest req = mapping.request();
- for (KeyCacheObject k : keys) {
- IgniteBiTuple<GridCacheVersion, CacheObject> oldValTup = valMap.get(k);
+ assert req != null : mapping;
- CacheObject newVal = res.value(i);
+ req.topologyVersion(res.clientRemapVersion());
+ }
+ }
+ }
+ }
+ else {
+ IgniteInternalFuture<?> affFut =
+ cctx.shared().exchange().affinityReadyFuture(res.clientRemapVersion());
- GridCacheVersion dhtVer = res.dhtVersion(i);
+ cctx.time().waitAsync(affFut, tx == null ? 0 : tx.remainingTime(), (e, timedOut) -> {
+ if (errorOrTimeoutOnTopologyVersion(e, timedOut))
+ return;
- if (newVal == null) {
- if (oldValTup != null) {
- if (oldValTup.get1().equals(dhtVer))
- newVal = oldValTup.get2();
+ try {
+ remap();
}
- }
+ finally {
+ cctx.shared().txContextReset();
+ }
+ });
- if (inTx()) {
- IgniteTxEntry txEntry = tx.entry(cctx.txKey(k));
+ return;
+ }
+ }
- // In colocated cache we must receive responses only for detached entries.
- assert txEntry.cached().detached() : txEntry;
+ int i = 0;
- txEntry.markLocked();
+ for (KeyCacheObject k : keys) {
+ IgniteBiTuple<GridCacheVersion, CacheObject> oldValTup = valMap.get(k);
- GridDhtDetachedCacheEntry entry = (GridDhtDetachedCacheEntry)txEntry.cached();
+ CacheObject newVal = res.value(i);
- if (res.dhtVersion(i) == null) {
- onDone(new IgniteCheckedException("Failed to receive DHT version from remote node " +
- "(will fail the lock): " + res));
+ GridCacheVersion dhtVer = res.dhtVersion(i);
- return;
- }
+ if (newVal == null) {
+ if (oldValTup != null) {
+ if (oldValTup.get1().equals(dhtVer))
+ newVal = oldValTup.get2();
+ }
+ }
- // Set value to detached entry.
- entry.resetFromPrimary(newVal, dhtVer);
+ if (inTx()) {
+ IgniteTxEntry txEntry = tx.entry(cctx.txKey(k));
- tx.hasRemoteLocks(true);
+ // In colocated cache we must receive responses only for detached entries.
+ assert txEntry.cached().detached() : txEntry;
- if (log.isDebugEnabled())
- log.debug("Processed response for entry [res=" + res + ", entry=" + entry + ']');
- }
- else
- cctx.mvcc().markExplicitOwner(cctx.txKey(k), threadId);
-
- if (retval && cctx.events().isRecordable(EVT_CACHE_OBJECT_READ)) {
- cctx.events().addEvent(cctx.affinity().partition(k),
- k,
- tx,
- null,
- EVT_CACHE_OBJECT_READ,
- newVal,
- newVal != null,
- null,
- false,
- CU.subjectId(tx, cctx.shared()),
- null,
- tx == null ? null : tx.resolveTaskName(),
- keepBinary);
+ txEntry.markLocked();
+
+ GridDhtDetachedCacheEntry entry = (GridDhtDetachedCacheEntry)txEntry.cached();
+
+ if (res.dhtVersion(i) == null) {
+ onDone(new IgniteCheckedException("Failed to receive DHT version from remote node " +
+ "(will fail the lock): " + res));
+
+ return;
}
- i++;
- }
+ // Set value to detached entry.
+ entry.resetFromPrimary(newVal, dhtVer);
- try {
- proceedMapping();
+ tx.hasRemoteLocks(true);
+
+ if (log.isDebugEnabled())
+ log.debug("Processed response for entry [res=" + res + ", entry=" + entry + ']');
}
- catch (IgniteCheckedException e) {
- onDone(e);
+ else
+ cctx.mvcc().markExplicitOwner(cctx.txKey(k), threadId);
+
+ if (retval && cctx.events().isRecordable(EVT_CACHE_OBJECT_READ)) {
+ cctx.events().addEvent(cctx.affinity().partition(k),
+ k,
+ tx,
+ null,
+ EVT_CACHE_OBJECT_READ,
+ newVal,
+ newVal != null,
+ null,
+ false,
+ CU.subjectId(tx, cctx.shared()),
+ null,
+ tx == null ? null : tx.resolveTaskName(),
+ keepBinary);
}
- onDone(true);
+ i++;
+ }
+
+ try {
+ proceedMapping();
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
}
+
+ onDone(true);
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
index 1b471f3..9061de0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
@@ -642,9 +642,9 @@ public class GridDhtPartitionDemander {
final RebalanceFuture fut = rebalanceFut;
- try {
- fut.cancelLock.readLock().lock();
+ fut.cancelLock.readLock().lock();
+ try {
ClusterNode node = ctx.node(nodeId);
if (node == null) {
@@ -1266,7 +1266,7 @@ public class GridDhtPartitionDemander {
/**
* @return Is initial (created at demander creation).
*/
- private boolean isInitial() {
+ public boolean isInitial() {
return topVer == null;
}
@@ -1276,11 +1276,11 @@ public class GridDhtPartitionDemander {
* @return {@code True}.
*/
@Override public boolean cancel() {
- try {
- // Cancel lock is needed only for case when some message might be on the fly while rebalancing is
- // cancelled.
- cancelLock.writeLock().lock();
+ // Cancel lock is needed only for case when some message might be on the fly while rebalancing is
+ // cancelled.
+ cancelLock.writeLock().lock();
+ try {
synchronized (this) {
if (isDone())
return true;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index 8fd8eb3..27cc0be 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -205,6 +205,9 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
/** */
private AtomicBoolean added = new AtomicBoolean(false);
+ /** Exchange type. */
+ private volatile ExchangeType exchangeType;
+
/**
* Discovery event receive latch. There is a race between discovery event processing and single message processing,
* so it is possible to create an exchange future before the actual discovery event is received. This latch is
@@ -353,8 +356,8 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
/** Discovery lag / Clocks discrepancy, calculated on coordinator when all single messages are received. */
private T2<Long, UUID> discoveryLag;
- /** Partitions scheduled for historical reblanace for this topology version. */
- private Map<Integer, Set<Integer>> histPartitions;
+ /** Partitions scheduled for clearing before rebalance for this topology version. */
+ private Map<Integer, Set<Integer>> clearingPartitions;
/**
* @param cctx Cache context.
@@ -499,6 +502,13 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
}
/**
+ * @return Exchange type or <code>null</code> if not determined yet.
+ */
+ public @Nullable ExchangeType exchangeType() {
+ return exchangeType;
+ }
+
+ /**
* Retreives the node which has WAL history since {@code cntrSince}.
*
* @param grpId Cache group ID.
@@ -857,6 +867,8 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
cctx.cache().registrateProxyRestart(resolveCacheRequests(exchActions), afterLsnrCompleteFut);
+ exchangeType = exchange;
+
updateTopologies(crdNode);
timeBag.finishGlobalStage("Determine exchange type");
@@ -1441,7 +1453,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
cctx.exchange().exchangerBlockingSectionEnd();
}
- histPartitions = new HashMap();
+ clearingPartitions = new HashMap();
timeBag.finishGlobalStage("WAL history reservation");
@@ -5080,31 +5092,35 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
/**
* @param grp Group.
* @param part Partition.
+ * @return {@code True} if partition has to be cleared before rebalance.
*/
- public boolean isHistoryPartition(CacheGroupContext grp, int part) {
+ public boolean isClearingPartition(CacheGroupContext grp, int part) {
if (!grp.persistenceEnabled())
return false;
synchronized (mux) {
- if (histPartitions == null)
+ if (clearingPartitions == null)
return false;
- Set<Integer> parts = histPartitions.get(grp.groupId());
+ Set<Integer> parts = clearingPartitions.get(grp.groupId());
return parts != null && parts.contains(part);
}
}
/**
+ * Marks a partition for clearing before rebalance.
+ * Fully cleared partitions should never be historically rebalanced.
+ *
* @param grp Group.
* @param part Partition.
*/
- public void addHistoryPartition(CacheGroupContext grp, int part) {
+ public void addClearingPartition(CacheGroupContext grp, int part) {
if (!grp.persistenceEnabled())
return;
synchronized (mux) {
- histPartitions.computeIfAbsent(grp.groupId(), k -> new HashSet()).add(part);
+ clearingPartitions.computeIfAbsent(grp.groupId(), k -> new HashSet()).add(part);
}
}
@@ -5144,7 +5160,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
/**
*
*/
- enum ExchangeType {
+ public enum ExchangeType {
/** */
CLIENT,
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
index 6cafd40..dd4c259 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
@@ -252,7 +252,7 @@ public class GridDhtPreloader extends GridCachePreloaderAdapter {
histSupplier = ctx.discovery().node(nodeId);
}
- if (histSupplier != null && exchFut.isHistoryPartition(grp, p)) {
+ if (histSupplier != null && !exchFut.isClearingPartition(grp, p)) {
assert grp.persistenceEnabled();
assert remoteOwners(p, topVer).contains(histSupplier) : remoteOwners(p, topVer);
@@ -270,6 +270,11 @@ public class GridDhtPreloader extends GridCachePreloaderAdapter {
msg.partitions().addHistorical(p, part.initialUpdateCounter(), countersMap.updateCounter(p), partitions);
}
else {
+ // If for some reason (for example if supplier fails and new supplier is elected) partition is
+ // assigned for full rebalance force clearing if not yet set.
+ if (grp.persistenceEnabled() && exchFut != null && !exchFut.isClearingPartition(grp, p))
+ part.clearAsync();
+
List<ClusterNode> picked = remoteOwners(p, topVer);
if (picked.isEmpty()) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
index 5260ab8..65447c8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
@@ -164,9 +164,6 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
* reservation is released. */
private volatile boolean delayedRenting;
- /** Set if partition must be cleared in MOVING state. */
- private volatile boolean clear;
-
/** Set if topology update sequence should be updated on partition destroy. */
private boolean updateSeqOnDestroy;
@@ -488,7 +485,7 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
if (reservations == 0)
return;
- assert getPartState(state) != EVICTED : getPartState(state);
+ assert getPartState(state) != EVICTED : this;
long newState = setReservations(state, --reservations);
newState = setSize(newState, getSize(newState) + sizeChange);
@@ -498,12 +495,14 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
// Decrement reservations.
if (this.state.compareAndSet(state, newState)) {
// If no more reservations try to continue delayed renting.
- if (reservations == 0 && delayedRenting)
- rent(true);
+ if (reservations == 0) {
+ if (delayedRenting)
+ rent(true);
+ else if (getPartState(state) == RENTING)
+ tryContinueClearing();
+ }
- // Partition could be only reserved in OWNING state so no further actions
- // are required.
- break;
+ return;
}
}
}
@@ -551,6 +550,8 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
boolean update = this.state.compareAndSet(state, setPartState(state, toState));
if (update) {
+ assert toState != EVICTED || reservations() == 0 : this;
+
try {
ctx.wal().log(new PartitionMetaStateRecord(grp.groupId(), id, toState, 0));
}
@@ -574,6 +575,8 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
boolean update = this.state.compareAndSet(state, setPartState(state, toState));
if (update) {
+ assert toState != EVICTED || reservations() == 0 : this;
+
if (log.isDebugEnabled())
log.debug("Partition changed state [grp=" + grp.cacheOrGroupName()
+ ", p=" + id + ", prev=" + prevState + ", to=" + toState + "]");
@@ -591,7 +594,6 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
long state = this.state.get();
GridDhtPartitionState partState = getPartState(state);
-
if (partState == RENTING || partState == EVICTED)
return false;
@@ -679,18 +681,43 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
GridDhtPartitionState partState = getPartState(state);
- boolean evictionRequested = partState == RENTING || delayedRenting;
- boolean clearingRequested = partState == MOVING && clear;
+ boolean evictionRequested = partState == RENTING;
+ boolean clearingRequested = partState == MOVING;
if (!evictionRequested && !clearingRequested)
return;
boolean reinitialized = clearFuture.initialize(updateSeq, evictionRequested);
- // Clearing process is already running at the moment. No needs to run it again.
+ // Clearing process is already running at the moment. No need to run it again.
if (!reinitialized)
return;
+ // Make sure current rebalance future is finished before start clearing
+ // to avoid clearing currently rebalancing partition (except "initial" dummy rebalance).
+ if (clearingRequested) {
+ GridDhtPartitionDemander.RebalanceFuture rebFut =
+ (GridDhtPartitionDemander.RebalanceFuture)grp.preloader().rebalanceFuture();
+
+ if (!rebFut.isInitial() && !rebFut.isDone()) {
+ rebFut.listen(fut -> {
+ // Partition could be owned after rebalance future is done. Skip clearing in such case.
+ // Otherwise continue clearing.
+ if (fut.error() == null && state() == MOVING) {
+ if (freeAndEmpty(state) && !grp.queriesEnabled() && !groupReserved()) {
+ clearFuture.finish();
+
+ return;
+ }
+
+ ctx.evict().evictPartitionAsync(grp, GridDhtLocalPartition.this);
+ }
+ });
+
+ return;
+ }
+ }
+
// Try fast eviction.
if (freeAndEmpty(state) && !grp.queriesEnabled() && !groupReserved()) {
if (partState == RENTING && casState(state, EVICTED) || clearingRequested) {
@@ -702,7 +729,7 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
destroy();
}
- if (log.isDebugEnabled())
+ if (log.isDebugEnabled() && evictionRequested)
log.debug("Partition has been fast evicted [grp=" + grp.cacheOrGroupName()
+ ", p=" + id + ", state=" + state() + "]");
@@ -716,6 +743,9 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
/**
* Initiates single clear process if partition is in MOVING state or continues cleaning for RENTING state.
* Method does nothing if clear process is already running.
+ *
+ * IMPORTANT: if clearing is required when after return from method call clear future must be initialized.
+ * This enforces clearing happens before sending demand requests.
*/
public void clearAsync() {
GridDhtPartitionState state0 = state();
@@ -723,18 +753,7 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
if (state0 != MOVING && state0 != RENTING)
return;
- clear = true;
-
- GridDhtPartitionDemander.RebalanceFuture rebFut =
- (GridDhtPartitionDemander.RebalanceFuture)grp.preloader().rebalanceFuture();
-
- // Make sure current rebalance future finishes before clearing
- // to avoid clearing currently rebalancing partition.
- // NOTE: this invariant is not true for initial rebalance future.
- if (rebFut.topologyVersion() != null && state0 == MOVING && !rebFut.isDone())
- rebFut.listen(fut -> clearAsync0(false));
- else
- clearAsync0(false);
+ clearAsync0(false);
}
/**
@@ -827,7 +846,10 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
GridDhtPartitionState state = getPartState(state0);
- if (state == EVICTED || (freeAndEmpty(state0) && state == RENTING && casState(state0, EVICTED)))
+ // Some entries still might be present in partition cache maps due to concurrent updates on backup nodes,
+ // but it's safe to finish eviction because no physical updates are possible.
+ if (state == EVICTED ||
+ (store.isEmpty() && getReservations(state0) == 0 && state == RENTING && casState(state0, EVICTED)))
updateSeqOnDestroy = updateSeq;
}
@@ -1132,7 +1154,7 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
// Do not clear fresh rows in case of partition reloading.
// This is required because normal updates are possible to moving partition which is currently cleared.
- if (row.version().compareTo(clearVer) >= 0 && (state() == MOVING && clear))
+ if (row.version().compareTo(clearVer) >= 0 && state() == MOVING)
continue;
if (grp.sharedGroup() && (hld == null || hld.cctx.cacheId() != row.cacheId()))
@@ -1302,6 +1324,7 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
"reservations", reservations(),
"empty", isEmpty(),
"createTime", U.format(createTime),
+ "fullSize", fullSize(),
"cntr", dataStore().partUpdateCounter());
}
@@ -1581,8 +1604,6 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
// Recreate cache data store in case of allowed fast eviction, and reset clear flag.
listen(f -> {
- clear = false;
-
clearingCbRegistered = false;
});
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java
index 3c9fa6e..e5d5651 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java
@@ -66,7 +66,6 @@ import org.apache.ignite.internal.util.tostring.GridToStringExclude;
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;
-import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.SB;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -75,6 +74,7 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.events.EventType.EVT_CACHE_REBALANCE_PART_DATA_LOST;
import static org.apache.ignite.internal.events.DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture.ExchangeType.ALL;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.EVICTED;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.LOST;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.MOVING;
@@ -766,7 +766,8 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
long updateSeq = this.updateSeq.incrementAndGet();
- if (!ctx.localNode().isClient()) {
+ // Skip partition updates in case of not real exchange.
+ if (!ctx.localNode().isClient() && exchFut.exchangeType() == ALL) {
for (int p = 0; p < partitions; p++) {
GridDhtLocalPartition locPart = localPartition0(p, topVer, false, true);
@@ -796,9 +797,10 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
// Node1 removes k2 but update has not been delivered to Node1 because of failure.
// After new full rebalance Node1 will only send k1 to Node2 causing lost removal.
// NOTE: avoid calling clearAsync for partition twice per topology version.
- // TODO FIXME clearing is not always needed see IGNITE-11799
- if (grp.persistenceEnabled() && !exchFut.isHistoryPartition(grp, locPart.id()) &&
- !locPart.isClearing() && !locPart.isEmpty() && !grp.mvccEnabled())
+ if (grp.persistenceEnabled() &&
+ exchFut.isClearingPartition(grp, locPart.id()) &&
+ !locPart.isClearing() &&
+ !locPart.isEmpty())
locPart.clearAsync();
}
else
@@ -871,7 +873,7 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
* @param p Partition number.
* @return Partition.
*/
- private GridDhtLocalPartition getOrCreatePartition(int p) {
+ public GridDhtLocalPartition getOrCreatePartition(int p) {
assert lock.isWriteLockedByCurrentThread();
assert ctx.database().checkpointLockIsHeldByThread();
@@ -1171,9 +1173,11 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
AffinityTopologyVersion diffVer = diffFromAffinityVer;
if (!diffVer.equals(topVer)) {
- LT.warn(log, "Requested topology version does not match calculated diff, need to check if " +
- "affinity has changed [grp=" + grp.cacheOrGroupName() + ", topVer=" + topVer +
- ", diffVer=" + diffVer + "]");
+ if (log.isDebugEnabled()) {
+ log.debug("Requested topology version does not match calculated diff, need to check if " +
+ "affinity has changed [grp=" + grp.cacheOrGroupName() + ", topVer=" + topVer +
+ ", diffVer=" + diffVer + "]");
+ }
boolean affChanged;
@@ -1183,9 +1187,11 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
affChanged = ctx.exchange().affinityChanged(topVer, diffVer);
if (affChanged) {
- LT.warn(log, "Requested topology version does not match calculated diff, will require full iteration to" +
- "calculate mapping [grp=" + grp.cacheOrGroupName() + ", topVer=" + topVer +
- ", diffVer=" + diffVer + "]");
+ if (log.isDebugEnabled()) {
+ log.debug("Requested topology version does not match calculated diff, will require full iteration to" +
+ "calculate mapping [grp=" + grp.cacheOrGroupName() + ", topVer=" + topVer +
+ ", diffVer=" + diffVer + "]");
+ }
nodes = new ArrayList<>();
@@ -1616,7 +1622,10 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
}
}
else if (state == MOVING) {
- rebalancePartition(p, partsToReload.contains(p), exchFut);
+ GridDhtLocalPartition locPart = locParts.get(p);
+
+ rebalancePartition(p, partsToReload.contains(p) ||
+ locPart != null && locPart.state() == MOVING && exchFut.localJoinExchange(), exchFut);
changed = true;
}
@@ -2368,8 +2377,8 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
if (part.state() != MOVING)
part.moving();
- if (!clear)
- exchFut.addHistoryPartition(grp, part.id());
+ if (clear)
+ exchFut.addClearingPartition(grp, part.id());
assert part.state() == MOVING : part;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/PartitionsEvictManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/PartitionsEvictManager.java
index 826902c..3c69905 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/PartitionsEvictManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/PartitionsEvictManager.java
@@ -369,7 +369,7 @@ public class PartitionsEvictManager extends GridCacheSharedManagerAdapter {
", grpId=" + grp.groupId() +
", remainingPartsToEvict=" + (totalTasks.get() - taskInProgress) +
", partsEvictInProgress=" + taskInProgress +
- ", totalParts= " + grp.topology().localPartitions().size() + "]");
+ ", totalParts=" + grp.topology().localPartitions().size() + "]");
}
}
@@ -412,12 +412,8 @@ public class PartitionsEvictManager extends GridCacheSharedManagerAdapter {
}
try {
- assert part.state() != GridDhtPartitionState.OWNING : part;
-
boolean success = part.tryClear(grpEvictionCtx);
- assert part.state() != GridDhtPartitionState.OWNING : part;
-
if (success) {
if (part.state() == GridDhtPartitionState.EVICTED && part.markForDestroy())
part.destroy();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
index 219fe2a..97535f5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
@@ -917,7 +917,7 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
if (remap) {
if (tx != null)
- tx.onRemap(topVer);
+ tx.onRemap(topVer, true);
this.topVer = topVer;
}
@@ -1645,143 +1645,167 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
if (res.clientRemapVersion() != null) {
assert cctx.kernalContext().clientNode();
- IgniteInternalFuture<?> affFut =
- cctx.shared().exchange().affinityReadyFuture(res.clientRemapVersion());
+ if (res.compatibleRemapVersion()) {
+ if (tx != null) {
+ // Versions are compatible.
+ cctx.shared().exchange().
+ lastAffinityChangedTopologyVersion(res.clientRemapVersion(), tx.topologyVersionSnapshot());
- if (affFut != null && !affFut.isDone()) {
- affFut.listen(new CI1<IgniteInternalFuture<?>>() {
- @Override public void apply(IgniteInternalFuture<?> fut) {
- try {
- fut.get();
+ tx.onRemap(res.clientRemapVersion(), false);
- remap();
- }
- catch (IgniteCheckedException e) {
- onDone(e);
- }
- finally {
- cctx.shared().txContextReset();
+ // Use remapped version for all subsequent mappings.
+ synchronized (GridNearLockFuture.this) {
+ for (GridNearLockMapping mapping : mappings) {
+ GridNearLockRequest req = mapping.request();
+
+ assert req != null : mapping;
+
+ req.topologyVersion(res.clientRemapVersion());
}
}
- });
+ }
}
- else
- remap();
- }
- else {
- int i = 0;
+ else {
+ IgniteInternalFuture<?> affFut =
+ cctx.shared().exchange().affinityReadyFuture(res.clientRemapVersion());
- AffinityTopologyVersion topVer = GridNearLockFuture.this.topVer;
+ if (!affFut.isDone()) {
+ // TODO FIXME https://ggsystems.atlassian.net/browse/GG-23288
+ affFut.listen(new CI1<IgniteInternalFuture<?>>() {
+ @Override public void apply(IgniteInternalFuture<?> fut) {
+ try {
+ fut.get();
- for (KeyCacheObject k : keys) {
- while (true) {
- GridNearCacheEntry entry = cctx.near().entryExx(k, topVer);
+ remap();
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
+ finally {
+ cctx.shared().txContextReset();
+ }
+ }
+ });
+ }
+ else
+ remap();
- try {
- if (res.dhtVersion(i) == null) {
- onDone(new IgniteCheckedException("Failed to receive DHT version from remote node " +
- "(will fail the lock): " + res));
+ return;
+ }
+ }
- return;
- }
+ int i = 0;
- IgniteBiTuple<GridCacheVersion, CacheObject> oldValTup = valMap.get(entry.key());
+ AffinityTopologyVersion topVer = GridNearLockFuture.this.topVer;
- CacheObject oldVal = entry.rawGet();
- boolean hasOldVal = false;
- CacheObject newVal = res.value(i);
+ for (KeyCacheObject k : keys) {
+ while (true) {
+ GridNearCacheEntry entry = cctx.near().entryExx(k, topVer);
- boolean readRecordable = false;
+ try {
+ if (res.dhtVersion(i) == null) {
+ onDone(new IgniteCheckedException("Failed to receive DHT version from remote node " +
+ "(will fail the lock): " + res));
- if (retval) {
- readRecordable = cctx.events().isRecordable(EVT_CACHE_OBJECT_READ);
+ return;
+ }
- if (readRecordable)
- hasOldVal = entry.hasValue();
- }
+ IgniteBiTuple<GridCacheVersion, CacheObject> oldValTup = valMap.get(entry.key());
- GridCacheVersion dhtVer = res.dhtVersion(i);
- GridCacheVersion mappedVer = res.mappedVersion(i);
+ CacheObject oldVal = entry.rawGet();
+ boolean hasOldVal = false;
+ CacheObject newVal = res.value(i);
- if (newVal == null) {
- if (oldValTup != null) {
- if (oldValTup.get1().equals(dhtVer))
- newVal = oldValTup.get2();
+ boolean readRecordable = false;
- oldVal = oldValTup.get2();
- }
- }
+ if (retval) {
+ readRecordable = cctx.events().isRecordable(EVT_CACHE_OBJECT_READ);
- // Lock is held at this point, so we can set the
- // returned value if any.
- entry.resetFromPrimary(newVal, lockVer, dhtVer, node.id(), topVer);
+ if (readRecordable)
+ hasOldVal = entry.hasValue();
+ }
- if (inTx()) {
- tx.hasRemoteLocks(true);
+ GridCacheVersion dhtVer = res.dhtVersion(i);
+ GridCacheVersion mappedVer = res.mappedVersion(i);
- if (implicitTx() && tx.onePhaseCommit()) {
- boolean pass = res.filterResult(i);
+ if (newVal == null) {
+ if (oldValTup != null) {
+ if (oldValTup.get1().equals(dhtVer))
+ newVal = oldValTup.get2();
- tx.entry(cctx.txKey(k)).filters(pass ? CU.empty0() : CU.alwaysFalse0Arr());
- }
+ oldVal = oldValTup.get2();
}
+ }
- entry.readyNearLock(lockVer,
- mappedVer,
- res.committedVersions(),
- res.rolledbackVersions(),
- res.pending());
-
- if (retval) {
- if (readRecordable)
- cctx.events().addEvent(
- entry.partition(),
- entry.key(),
- tx,
- null,
- EVT_CACHE_OBJECT_READ,
- newVal,
- newVal != null,
- oldVal,
- hasOldVal,
- CU.subjectId(tx, cctx.shared()),
- null,
- inTx() ? tx.resolveTaskName() : null,
- keepBinary);
-
- if (cctx.statisticsEnabled())
- cctx.cache().metrics0().onRead(false);
- }
+ // Lock is held at this point, so we can set the
+ // returned value if any.
+ entry.resetFromPrimary(newVal, lockVer, dhtVer, node.id(), topVer);
- if (log.isDebugEnabled())
- log.debug("Processed response for entry [res=" + res + ", entry=" + entry + ']');
+ if (inTx()) {
+ tx.hasRemoteLocks(true);
- break; // Inner while loop.
- }
- catch (GridCacheEntryRemovedException ignored) {
- if (log.isDebugEnabled())
- log.debug("Failed to add candidates because entry was removed (will renew).");
+ if (implicitTx() && tx.onePhaseCommit()) {
+ boolean pass = res.filterResult(i);
- synchronized (GridNearLockFuture.this) {
- // Replace old entry with new one.
- entries.set(i,
- (GridDistributedCacheEntry)cctx.cache().entryEx(entry.key()));
+ tx.entry(cctx.txKey(k)).filters(pass ? CU.empty0() : CU.alwaysFalse0Arr());
}
}
+
+ entry.readyNearLock(lockVer,
+ mappedVer,
+ res.committedVersions(),
+ res.rolledbackVersions(),
+ res.pending());
+
+ if (retval) {
+ if (readRecordable)
+ cctx.events().addEvent(
+ entry.partition(),
+ entry.key(),
+ tx,
+ null,
+ EVT_CACHE_OBJECT_READ,
+ newVal,
+ newVal != null,
+ oldVal,
+ hasOldVal,
+ CU.subjectId(tx, cctx.shared()),
+ null,
+ inTx() ? tx.resolveTaskName() : null,
+ keepBinary);
+
+ if (cctx.statisticsEnabled())
+ cctx.cache().metrics0().onRead(false);
+ }
+
+ if (log.isDebugEnabled())
+ log.debug("Processed response for entry [res=" + res + ", entry=" + entry + ']');
+
+ break; // Inner while loop.
}
+ catch (GridCacheEntryRemovedException ignored) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to add candidates because entry was removed (will renew).");
- i++;
+ synchronized (GridNearLockFuture.this) {
+ // Replace old entry with new one.
+ entries.set(i,
+ (GridDistributedCacheEntry)cctx.cache().entryEx(entry.key()));
+ }
+ }
}
- try {
- proceedMapping();
- }
- catch (IgniteCheckedException e) {
- onDone(e);
- }
+ i++;
+ }
- onDone(true);
+ try {
+ proceedMapping();
}
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
+
+ onDone(true);
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
index ce913aa..86709b7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
@@ -226,6 +226,13 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
}
/**
+ * @param topVer Topology version.
+ */
+ public void topologyVersion(AffinityTopologyVersion topVer) {
+ this.topVer = topVer;
+ }
+
+ /**
* @return Subject ID.
*/
public UUID subjectId() {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
index b6c6d8c..21d0232 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
@@ -60,9 +60,12 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
/** Filter evaluation results for fast-commit transactions. */
private boolean[] filterRes;
- /** {@code True} if client node should remap lock request. */
+ /** Set if client node should remap lock request. */
private AffinityTopologyVersion clientRemapVer;
+ /** {@code True} if remap version is compatible with current version. Used together with clientRemapVer. */
+ private boolean compatibleRemapVer;
+
/**
* Empty constructor (required by {@link Externalizable}).
*/
@@ -78,8 +81,10 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
* @param filterRes {@code True} if need to allocate array for filter evaluation results.
* @param cnt Count.
* @param err Error.
- * @param clientRemapVer {@code True} if client node should remap lock request.
+ * @param clientRemapVer {@code True} if client node should remap lock request. If {@code compatibleRemapVer} is
+ * {@code true} when first request is not remapped, but all subsequent will use remap version.
* @param addDepInfo Deployment info.
+ * @param compatibleRemapVer {@code True} if remap version is compatible with lock version.
*/
public GridNearLockResponse(
int cacheId,
@@ -90,7 +95,8 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
int cnt,
Throwable err,
AffinityTopologyVersion clientRemapVer,
- boolean addDepInfo
+ boolean addDepInfo,
+ boolean compatibleRemapVer
) {
super(cacheId, lockVer, futId, cnt, err, addDepInfo);
@@ -104,6 +110,8 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
if (filterRes)
this.filterRes = new boolean[cnt];
+
+ this.compatibleRemapVer = compatibleRemapVer;
}
/**
@@ -114,6 +122,13 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
}
/**
+ * @return {@code True} is remap version is compatible with current topology version.
+ */
+ public boolean compatibleRemapVersion() {
+ return compatibleRemapVer;
+ }
+
+ /**
* Gets pending versions that are less than {@link #version()}.
*
* @return Pending versions.
@@ -215,30 +230,36 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
writer.incrementState();
case 12:
- if (!writer.writeObjectArray("dhtVers", dhtVers, MessageCollectionItemType.MSG))
+ if (!writer.writeBoolean("compatibleRemapVer", compatibleRemapVer))
return false;
writer.incrementState();
case 13:
- if (!writer.writeBooleanArray("filterRes", filterRes))
+ if (!writer.writeObjectArray("dhtVers", dhtVers, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 14:
- if (!writer.writeObjectArray("mappedVers", mappedVers, MessageCollectionItemType.MSG))
+ if (!writer.writeBooleanArray("filterRes", filterRes))
return false;
writer.incrementState();
case 15:
- if (!writer.writeInt("miniId", miniId))
+ if (!writer.writeObjectArray("mappedVers", mappedVers, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 16:
+ if (!writer.writeInt("miniId", miniId))
+ return false;
+
+ writer.incrementState();
+
+ case 17:
if (!writer.writeCollection("pending", pending, MessageCollectionItemType.MSG))
return false;
@@ -269,7 +290,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
reader.incrementState();
case 12:
- dhtVers = reader.readObjectArray("dhtVers", MessageCollectionItemType.MSG, GridCacheVersion.class);
+ compatibleRemapVer = reader.readBoolean("compatibleRemapVer");
if (!reader.isLastRead())
return false;
@@ -277,7 +298,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
reader.incrementState();
case 13:
- filterRes = reader.readBooleanArray("filterRes");
+ dhtVers = reader.readObjectArray("dhtVers", MessageCollectionItemType.MSG, GridCacheVersion.class);
if (!reader.isLastRead())
return false;
@@ -285,7 +306,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
reader.incrementState();
case 14:
- mappedVers = reader.readObjectArray("mappedVers", MessageCollectionItemType.MSG, GridCacheVersion.class);
+ filterRes = reader.readBooleanArray("filterRes");
if (!reader.isLastRead())
return false;
@@ -293,7 +314,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
reader.incrementState();
case 15:
- miniId = reader.readInt("miniId");
+ mappedVers = reader.readObjectArray("mappedVers", MessageCollectionItemType.MSG, GridCacheVersion.class);
if (!reader.isLastRead())
return false;
@@ -301,6 +322,14 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
reader.incrementState();
case 16:
+ miniId = reader.readInt("miniId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 17:
pending = reader.readCollection("pending", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
@@ -320,7 +349,7 @@ public class GridNearLockResponse extends GridDistributedLockResponse {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 17;
+ return 18;
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
index 791e018..e1606ba 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
@@ -62,7 +62,6 @@ import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiClosure;
-import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.transactions.TransactionDeadlockException;
import org.apache.ignite.transactions.TransactionTimeoutException;
import org.jetbrains.annotations.Nullable;
@@ -156,30 +155,8 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
}
}
- if (ERR_UPD.compareAndSet(this, null, e)) {
- boolean marked = tx.setRollbackOnly();
-
- if (e instanceof IgniteTxRollbackCheckedException) {
- if (marked) {
- tx.rollbackAsync().listen(new IgniteInClosure<IgniteInternalFuture<IgniteInternalTx>>() {
- @Override public void apply(IgniteInternalFuture<IgniteInternalTx> fut) {
- try {
- fut.get();
- }
- catch (IgniteCheckedException e) {
- U.error(log, "Failed to automatically rollback transaction: " + tx, e);
- }
-
- onComplete();
- }
- });
-
- return;
- }
- }
-
+ if (ERR_UPD.compareAndSet(this, null, e))
onComplete();
- }
}
/** {@inheritDoc} */
@@ -537,7 +514,7 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
tx.subjectId(),
tx.taskNameHash(),
m.clientFirst(),
- true,
+ txMapping.transactionNodes().size() == 1,
tx.activeCachesDeploymentEnabled(),
tx.txState().recovery());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
index fb62880..eee6612 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
@@ -156,7 +156,7 @@ public abstract class GridNearOptimisticTxPrepareFutureAdapter extends GridNearT
}
if (remap)
- tx.onRemap(topVer);
+ tx.onRemap(topVer, true);
else
tx.topologyVersion(topVer);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
index 7c60352..858682c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
@@ -4584,18 +4584,21 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
/**
* @param topVer New topology version.
+ * @param reset {@code True} if need to reset tx state.
*/
- public void onRemap(AffinityTopologyVersion topVer) {
+ public void onRemap(AffinityTopologyVersion topVer, boolean reset) {
assert cctx.kernalContext().clientNode();
- mapped = false;
- nearLocallyMapped = false;
- colocatedLocallyMapped = false;
- txNodes = null;
- onePhaseCommit = false;
- nearMap.clear();
- dhtMap.clear();
- mappings.clear();
+ if (reset) {
+ mapped = false;
+ nearLocallyMapped = false;
+ colocatedLocallyMapped = false;
+ txNodes = null;
+ onePhaseCommit = false;
+ nearMap.clear();
+ dhtMap.clear();
+ mappings.clear();
+ }
synchronized (this) {
this.topVer = topVer;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/PartitionCountersNeighborcastRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/PartitionCountersNeighborcastRequest.java
index 5c46ca6..c053e10 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/PartitionCountersNeighborcastRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/PartitionCountersNeighborcastRequest.java
@@ -20,12 +20,14 @@ package org.apache.ignite.internal.processors.cache.mvcc.msg;
import java.nio.ByteBuffer;
import java.util.Collection;
import org.apache.ignite.internal.GridDirectCollection;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheIdMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessage;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.jetbrains.annotations.NotNull;
/** */
public class PartitionCountersNeighborcastRequest extends GridCacheIdMessage {
@@ -39,15 +41,27 @@ public class PartitionCountersNeighborcastRequest extends GridCacheIdMessage {
/** */
private IgniteUuid futId;
+ /** Topology version. */
+ private AffinityTopologyVersion topVer;
+
/** */
public PartitionCountersNeighborcastRequest() {
}
/** */
public PartitionCountersNeighborcastRequest(
- Collection<PartitionUpdateCountersMessage> updCntrs, IgniteUuid futId) {
+ Collection<PartitionUpdateCountersMessage> updCntrs,
+ IgniteUuid futId,
+ @NotNull AffinityTopologyVersion topVer
+ ) {
this.updCntrs = updCntrs;
this.futId = futId;
+ this.topVer = topVer;
+ }
+
+ /** {@inheritDoc} */
+ @Override public AffinityTopologyVersion topologyVersion() {
+ return topVer;
}
/**
@@ -86,6 +100,12 @@ public class PartitionCountersNeighborcastRequest extends GridCacheIdMessage {
writer.incrementState();
case 5:
+ if (!writer.writeAffinityTopologyVersion("topVer", topVer))
+ return false;
+
+ writer.incrementState();
+
+ case 6:
if (!writer.writeCollection("updCntrs", updCntrs, MessageCollectionItemType.MSG))
return false;
@@ -116,6 +136,14 @@ public class PartitionCountersNeighborcastRequest extends GridCacheIdMessage {
reader.incrementState();
case 5:
+ topVer = reader.readAffinityTopologyVersion("topVer");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 6:
updCntrs = reader.readCollection("updCntrs", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
@@ -135,7 +163,7 @@ public class PartitionCountersNeighborcastRequest extends GridCacheIdMessage {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 6;
+ return 7;
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/PartitionCountersNeighborcastResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/PartitionCountersNeighborcastResponse.java
index 2f88cde..44a6949 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/PartitionCountersNeighborcastResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/msg/PartitionCountersNeighborcastResponse.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.cache.mvcc.msg;
import java.nio.ByteBuffer;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheIdMessage;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
@@ -31,13 +32,25 @@ public class PartitionCountersNeighborcastResponse extends GridCacheIdMessage {
/** */
private IgniteUuid futId;
+ /** Topology version. */
+ private AffinityTopologyVersion topVer;
+
/** */
public PartitionCountersNeighborcastResponse() {
}
/** */
- public PartitionCountersNeighborcastResponse(IgniteUuid futId) {
+ public PartitionCountersNeighborcastResponse(
+ IgniteUuid futId,
+ AffinityTopologyVersion topVer
+ ) {
this.futId = futId;
+ this.topVer = topVer;
+ }
+
+ /** {@inheritDoc} */
+ @Override public AffinityTopologyVersion topologyVersion() {
+ return topVer;
}
/**
@@ -68,6 +81,12 @@ public class PartitionCountersNeighborcastResponse extends GridCacheIdMessage {
writer.incrementState();
+ case 5:
+ if (!writer.writeAffinityTopologyVersion("topVer", topVer))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -92,6 +111,14 @@ public class PartitionCountersNeighborcastResponse extends GridCacheIdMessage {
reader.incrementState();
+ case 5:
+ topVer = reader.readAffinityTopologyVersion("topVer");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(PartitionCountersNeighborcastResponse.class);
@@ -104,7 +131,7 @@ public class PartitionCountersNeighborcastResponse extends GridCacheIdMessage {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 5;
+ return 6;
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
index be340c4..21518ee 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
@@ -1608,7 +1608,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
/**
* @param partId Partition.
- * @param exists {@code True} if store for this index exists.
+ * @param exists {@code True} if store exists.
*/
private GridCacheDataStore(int partId, boolean exists) {
this.partId = partId;
@@ -1644,6 +1644,9 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
/**
+ * @param checkExists If {@code true} data store won't be initialized if it doesn't exists
+ * (has non empty data file). This is an optimization for lazy store initialization on writes.
+ *
* @return Store delegate.
* @throws IgniteCheckedException If failed.
*/
@@ -2212,11 +2215,9 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
/** {@inheritDoc} */
@Override public void updateInitialCounter(long start, long delta) {
try {
- CacheDataStore delegate0 = init0(true);
-
- if (delegate0 == null)
- throw new IllegalStateException("Should be never called.");
+ CacheDataStore delegate0 = init0(false);
+ // Partition may not exists before recovery starts in case of recovering counters from RollbackRecord.
delegate0.updateInitialCounter(start, delta);
}
catch (IgniteCheckedException e) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
index 6ea17c2..e62665f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
@@ -2240,7 +2240,7 @@ public class IgniteTxHandler {
}
try {
- ctx.io().send(nodeId, new PartitionCountersNeighborcastResponse(req.futId()), SYSTEM_POOL);
+ ctx.io().send(nodeId, new PartitionCountersNeighborcastResponse(req.futId(), req.topologyVersion()), SYSTEM_POOL);
}
catch (ClusterTopologyCheckedException ignored) {
if (txRecoveryMsgLog.isDebugEnabled())
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index 06e5ac1..a146ad5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -453,7 +453,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
* pair (init, delta) values, where init - initial update counter, and delta - updates count made
* by current transaction for a given partition.
*/
- public void calculatePartitionUpdateCounters() {
+ public void calculatePartitionUpdateCounters() throws IgniteTxRollbackCheckedException {
TxCounters counters = txCounters(false);
if (counters != null && F.isEmpty(counters.updateCounters())) {
@@ -491,11 +491,31 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
GridDhtLocalPartition part = top.localPartition(p);
+ // Verify primary tx mapping.
// LOST state is possible if tx is started over LOST partition.
- assert part != null && (part.state() == OWNING || part.state() == LOST):
- part == null ?
- "map=" + top.partitionMap(false) + ", lost=" + top.lostPartitions() :
- "part=" + part.toString();
+ boolean valid = part != null &&
+ (part.state() == OWNING || part.state() == LOST) &&
+ part.primary(top.readyTopologyVersion());
+
+ if (!valid) {
+ // Local node is no longer primary for the partition, need to rollback a transaction.
+ if (part != null && !part.primary(top.readyTopologyVersion())) {
+ log.warning("Failed to prepare a transaction on outdated topology, rolling back " +
+ "[tx=" + CU.txString(this) +
+ ", readyTopVer=" + top.readyTopologyVersion() +
+ ", lostParts=" + top.lostPartitions() +
+ ", part=" + part.toString() + ']');
+
+ throw new IgniteTxRollbackCheckedException("Failed to prepare a transaction on outdated " +
+ "topology, please try again [timeout=" + timeout() + ", tx=" + CU.txString(this) + ']');
+ }
+
+ // Trigger error.
+ throw new AssertionError("Invalid primary mapping [tx=" + CU.txString(this) +
+ ", readyTopVer=" + top.readyTopologyVersion() +
+ ", lostParts=" + top.lostPartitions() +
+ ", part=" + (part == null ? "NULL" : part.toString()) + ']');
+ }
msg.add(p, part.getAndIncrementUpdateCounter(cntr), cntr);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/PartitionCountersNeighborcastFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/PartitionCountersNeighborcastFuture.java
index 93e3ef2..39b79d4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/PartitionCountersNeighborcastFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/PartitionCountersNeighborcastFuture.java
@@ -98,7 +98,7 @@ public class PartitionCountersNeighborcastFuture extends GridCacheCompoundIdenti
// we must add mini future before sending a message, otherwise mini future must miss completion
add(miniFut);
- cctx.io().send(peer, new PartitionCountersNeighborcastRequest(cntrs, futId), SYSTEM_POOL);
+ cctx.io().send(peer, new PartitionCountersNeighborcastRequest(cntrs, futId, tx.topologyVersion()), SYSTEM_POOL);
}
catch (IgniteCheckedException e) {
if (!(e instanceof ClusterTopologyCheckedException))
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/ResetLostPartitionTest.java b/modules/core/src/test/java/org/apache/ignite/cache/ResetLostPartitionTest.java
index ef74ba5..427d3a5 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/ResetLostPartitionTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/ResetLostPartitionTest.java
@@ -21,7 +21,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
@@ -32,14 +32,12 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.topology.Grid
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopologyImpl;
import org.apache.ignite.internal.util.typedef.internal.CU;
-import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
import static org.apache.ignite.configuration.WALMode.LOG_ONLY;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.LOST;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
-import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.DFLT_STORE_DIR;
/**
*
@@ -151,25 +149,18 @@ public class ResetLostPartitionTest extends GridCommonAbstractTest {
grid(0).cluster().active(true);
- Ignite igniteClient = startGrid(getClientConfiguration("client"));
-
for (String cacheName : CACHE_NAMES) {
- IgniteCache<Integer, String> cache = igniteClient.cache(cacheName);
-
- for (int j = 0; j < CACHE_SIZE; j++)
- cache.put(j, "Value" + j);
+ try(IgniteDataStreamer<Object, Object> st = grid(0).dataStreamer(cacheName)) {
+ for (int j = 0; j < CACHE_SIZE; j++)
+ st.addData(j, "Value" + j);
+ }
}
- stopGrid("client");
-
- String dn2DirName = grid(1).name().replace(".", "_");
+ String g1Name = grid(1).name();
stopGrid(1);
- //Clean up the pds and WAL for second data node.
- U.delete(U.resolveWorkDirectory(U.defaultWorkDirectory(), DFLT_STORE_DIR + "/" + dn2DirName, true));
-
- U.delete(U.resolveWorkDirectory(U.defaultWorkDirectory(), DFLT_STORE_DIR + "/wal/" + dn2DirName, true));
+ cleanPersistenceDir(g1Name);
//Here we have two from three data nodes and cache with 1 backup. So there is no data loss expected.
assertEquals(CACHE_NAMES.length * CACHE_SIZE, averageSizeAroundAllNodes());
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheDataLossOnPartitionMoveTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheDataLossOnPartitionMoveTest.java
index 60d8b42..cdf6c24 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheDataLossOnPartitionMoveTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheDataLossOnPartitionMoveTest.java
@@ -142,7 +142,7 @@ public class CacheDataLossOnPartitionMoveTest extends GridCommonAbstractTest {
ignite.cluster().active(true);
List<Integer> toCp = movingKeysAfterJoin(ignite, DEFAULT_CACHE_NAME, 1,
- node -> ((GridTestNode)node).setAttribute(GRP_ATTR, ODD_GRP));
+ node -> ((GridTestNode)node).setAttribute(GRP_ATTR, ODD_GRP), null);
int blockPartId = ignite.affinity(DEFAULT_CACHE_NAME).partition(toCp.get(0));
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CachePartitionLossDetectionOnNodeLeftTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CachePartitionLossDetectionOnNodeLeftTest.java
index 61b024c..2388907 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CachePartitionLossDetectionOnNodeLeftTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CachePartitionLossDetectionOnNodeLeftTest.java
@@ -29,8 +29,6 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
@@ -40,7 +38,6 @@ import static org.apache.ignite.testframework.GridTestUtils.mergeExchangeWaitVer
/**
* Tests if lost partitions are same on left nodes after other owners removal.
*/
-@RunWith(JUnit4.class)
public class CachePartitionLossDetectionOnNodeLeftTest extends GridCommonAbstractTest {
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CachePartitionLostWhileClearingTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CachePartitionLostWhileClearingTest.java
new file mode 100644
index 0000000..53cb27b
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CachePartitionLostWhileClearingTest.java
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed;
+
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.WALMode;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_BASELINE_AUTO_ADJUST_ENABLED;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+
+/**
+ * Test scenario: supplier is left during rebalancing leaving partition in OWNING state (but actually LOST)
+ * because only one owner left.
+ * <p>
+ * Expected result: no assertions are triggered.
+ */
+@WithSystemProperty(key = IGNITE_BASELINE_AUTO_ADJUST_ENABLED, value = "false")
+public class CachePartitionLostWhileClearingTest extends GridCommonAbstractTest {
+ /** */
+ private static final int PARTS_CNT = 64;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setConsistentId(igniteInstanceName);
+
+ cfg.setDataStorageConfiguration(
+ new DataStorageConfiguration()
+ .setWalMode(WALMode.LOG_ONLY)
+ .setWalSegmentSize(4 * 1024 * 1024)
+ .setDefaultDataRegionConfiguration(
+ new DataRegionConfiguration()
+ .setPersistenceEnabled(true)
+ .setMaxSize(100L * 1024 * 1024))
+ );
+
+ cfg.setCacheConfiguration(new CacheConfiguration(DEFAULT_CACHE_NAME).
+ setAtomicityMode(TRANSACTIONAL).
+ setCacheMode(PARTITIONED).
+ setBackups(1).
+ setAffinity(new RendezvousAffinityFunction(false, PARTS_CNT)));
+
+ return cfg;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ cleanPersistenceDir();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ cleanPersistenceDir();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testPartitionLostWhileClearing_FailOnCrd() throws Exception {
+ doTestPartitionLostWhileClearing(2);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testPartitionLostWhileClearing_FailOnFullMessage() throws Exception {
+ doTestPartitionLostWhileClearing(3);
+ }
+
+ /**
+ * @param cnt Nodes count.
+ *
+ * @throws Exception If failed.
+ */
+ private void doTestPartitionLostWhileClearing(int cnt) throws Exception {
+ try {
+ IgniteEx crd = startGrids(cnt);
+
+ crd.cluster().active(true);
+
+ int partId = -1;
+ int idx0 = 0;
+ int idx1 = 1;
+
+ for (int p = 0; p < PARTS_CNT; p++) {
+ List<ClusterNode> nodes = new ArrayList<>(crd.affinity(DEFAULT_CACHE_NAME).mapPartitionToPrimaryAndBackups(p));
+
+ if (grid(nodes.get(0)) == grid(idx0) && grid(nodes.get(1)) == grid(idx1)) {
+ partId = p;
+
+ break;
+ }
+ }
+
+ assertTrue(partId >= 0);
+
+ load(grid(idx0), DEFAULT_CACHE_NAME, partId, 10_000, 0);
+
+ stopGrid(idx1);
+
+ load(grid(idx0), DEFAULT_CACHE_NAME, partId, 10, 10_000);
+
+ IgniteEx g1 = startGrid(idx1);
+
+ stopGrid(idx0); // Restart supplier in the middle of clearing.
+
+ awaitPartitionMapExchange();
+
+ // Expecting partition in OWNING state.
+ assertNotNull(counter(partId, DEFAULT_CACHE_NAME, g1.name()));
+ }
+ finally {
+ stopAllGrids();
+ }
+ }
+
+ /**
+ * @param ignite Ignite.
+ * @param cache Cache.
+ * @param partId Partition id.
+ * @param cnt Count.
+ * @param skip Skip.
+ */
+ private void load(IgniteEx ignite, String cache, int partId, int cnt, int skip) {
+ List<Integer> keys = partitionKeys(ignite.cache(cache), partId, cnt, skip);
+
+ try(IgniteDataStreamer<Object, Object> s = ignite.dataStreamer(cache)) {
+ for (Integer key : keys)
+ s.addData(key, key);
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/GridCacheRebalancingSyncSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/GridCacheRebalancingSyncSelfTest.java
index 8d7ba7c..8af5747 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/GridCacheRebalancingSyncSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/GridCacheRebalancingSyncSelfTest.java
@@ -662,4 +662,9 @@ public class GridCacheRebalancingSyncSelfTest extends GridCommonAbstractTest {
super.sendMessage(node, msg, ackC);
}
}
+
+ /** {@inheritDoc} */
+ @Override protected long getPartitionMapExchangeTimeout() {
+ return super.getPartitionMapExchangeTimeout() * 2;
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/LocalWalModeChangeDuringRebalancingSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/LocalWalModeChangeDuringRebalancingSelfTest.java
index fb99aa8..1bb19cd 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/LocalWalModeChangeDuringRebalancingSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/LocalWalModeChangeDuringRebalancingSelfTest.java
@@ -34,11 +34,13 @@ import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.WALMode;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
@@ -98,15 +100,16 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr
.setInitialSize(DataStorageConfiguration.DFLT_DATA_REGION_INITIAL_SIZE)
.setMaxSize(DataStorageConfiguration.DFLT_DATA_REGION_INITIAL_SIZE)
)
- // Test verifies checkpoint count, so it is essencial that no checkpoint is triggered by timeout
+ // Test verifies checkpoint count, so it is essencial that no checkpoint is triggered by timeout
.setCheckpointFrequency(999_999_999_999L)
+ .setWalMode(WALMode.LOG_ONLY)
.setFileIOFactory(new TestFileIOFactory(new DataStorageConfiguration().getFileIOFactory()))
);
cfg.setCacheConfiguration(
new CacheConfiguration(DEFAULT_CACHE_NAME)
.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
- // Test checks internal state before and after rebalance, so it is configured to be triggered manually
+ // Test checks internal state before and after rebalance, so it is configured to be triggered manually
.setRebalanceDelay(-1)
.setBackups(dfltCacheBackupCnt),
@@ -340,7 +343,9 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr
awaitExchange((IgniteEx)ignite);
- doLoad(cache, 4, 10_000);
+ // Ensure each partition has received an update.
+ for (int k = 0; k < RendezvousAffinityFunction.DFLT_PARTITION_COUNT; k++)
+ cache.put(k, k);
IgniteEx newIgnite = startGrid(2);
@@ -445,7 +450,11 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr
for (int nodeIdx = 2; nodeIdx < nodeCnt; nodeIdx++) {
CacheGroupContext grpCtx = grid(nodeIdx).cachex(REPL_CACHE).context().group();
- assertFalse(grpCtx.walEnabled());
+ assertTrue(GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return !grpCtx.walEnabled();
+ }
+ }, 5_000));
}
// Invoke rebalance manually.
@@ -813,4 +822,9 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr
delegate.close();
}
}
+
+ /** {@inheritDoc} */
+ @Override protected long getPartitionMapExchangeTimeout() {
+ return super.getPartitionMapExchangeTimeout() * 2;
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/IgnitePdsCacheWalDisabledOnRebalancingTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/IgnitePdsCacheWalDisabledOnRebalancingTest.java
index 61861bc..6d80424 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/IgnitePdsCacheWalDisabledOnRebalancingTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/IgnitePdsCacheWalDisabledOnRebalancingTest.java
@@ -19,9 +19,7 @@ package org.apache.ignite.internal.processors.cache.persistence.db;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.nio.file.Files;
import java.nio.file.OpenOption;
-import java.nio.file.Paths;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
@@ -50,7 +48,6 @@ import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
import org.apache.ignite.internal.processors.cache.persistence.file.FileIODecorator;
import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory;
-import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.testframework.GridTestUtils;
@@ -58,7 +55,6 @@ import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_BASELINE_AUTO_ADJUST_ENABLED;
-import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.DFLT_STORE_DIR;
/**
* Test scenarios with rebalancing, IGNITE_DISABLE_WAL_DURING_REBALANCING optimization and topology changes
@@ -131,6 +127,9 @@ public class IgnitePdsCacheWalDisabledOnRebalancingTest extends GridCommonAbstra
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+ // This is required because some tests do full clearing of persistence folder losing BLT info on next join.
+ cfg.setConsistentId(igniteInstanceName);
+
CacheConfiguration ccfg1 = new CacheConfiguration("cache1")
.setAtomicityMode(CacheAtomicityMode.ATOMIC)
.setCacheMode(CacheMode.REPLICATED)
@@ -188,11 +187,11 @@ public class IgnitePdsCacheWalDisabledOnRebalancingTest extends GridCommonAbstra
for (int i = 1; i < 4; i++)
fillCache(ig0.dataStreamer("cache" + i), CACHE_SIZE, GENERATING_FUNC);
- String ig1Name = "node01-" + grid(1).localNode().consistentId();
+ String ig1Name = grid(1).name();
stopGrid(1);
- cleanPersistenceFiles(ig1Name);
+ cleanPersistenceDir(ig1Name);
int groupId = ((IgniteEx) ig0).cachex(CACHE3_NAME).context().groupId();
@@ -298,9 +297,9 @@ public class IgnitePdsCacheWalDisabledOnRebalancingTest extends GridCommonAbstra
stopGrid(2);
// Stop idx=1 and cleanup LFS to trigger full rebalancing after it restart.
- String ig1Name = "node01-" + grid(1).localNode().consistentId();
+ String ig1Name = grid(1).name();
stopGrid(1);
- cleanPersistenceFiles(ig1Name);
+ cleanPersistenceDir(ig1Name);
// Blocking fileIO and blockMessagePredicate to block checkpointer and rebalancing for node idx=1.
useBlockingFileIO = true;
@@ -356,6 +355,43 @@ public class IgnitePdsCacheWalDisabledOnRebalancingTest extends GridCommonAbstra
verifyCache(ig1.cache(CACHE3_NAME), GENERATING_FUNC);
}
+ /**
+ * Scenario: when rebalanced MOVING partitions are owning by checkpointer,
+ * concurrent no-op exchange should not trigger partition clearing.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testRebalancedPartitionsOwningWithAffinitySwitch() throws Exception {
+ Ignite ig0 = startGridsMultiThreaded(4);
+ fillCache(ig0.dataStreamer(CACHE3_NAME), CACHE_SIZE, GENERATING_FUNC);
+
+ // Stop idx=2 to prepare for baseline topology change later.
+ stopGrid(2);
+
+ // Stop idx=1 and cleanup LFS to trigger full rebalancing after it restart.
+ String ig1Name = grid(1).name();
+ stopGrid(1);
+ cleanPersistenceDir(ig1Name);
+
+ // Blocking fileIO and blockMessagePredicate to block checkpointer and rebalancing for node idx=1.
+ useBlockingFileIO = true;
+
+ // Enable blocking checkpointer on node idx=1 (see BlockingCheckpointFileIOFactory).
+ fileIoBlockingSemaphore.drainPermits();
+
+ // Wait for rebalance (all partitions will be in MOVING state until cp is finished).
+ startGrid(1).cachex(CACHE3_NAME).context().group().preloader().rebalanceFuture().get();
+
+ startGrid("client");
+
+ fileIoBlockingSemaphore.release(Integer.MAX_VALUE);
+
+ awaitPartitionMapExchange();
+
+ assertPartitionsSame(idleVerify(grid(0), CACHE3_NAME));
+ }
+
/** FileIOFactory implementation that enables blocking of writes to disk so checkpoint can be blocked. */
private static class BlockingCheckpointFileIOFactory implements FileIOFactory {
/** Serial version uid. */
@@ -412,20 +448,6 @@ public class IgnitePdsCacheWalDisabledOnRebalancingTest extends GridCommonAbstra
}
/** */
- private void cleanPersistenceFiles(String igName) throws Exception {
- String ig1DbPath = Paths.get(DFLT_STORE_DIR, igName).toString();
-
- File igDbDir = U.resolveWorkDirectory(U.defaultWorkDirectory(), ig1DbPath, false);
-
- U.delete(igDbDir);
- Files.createDirectory(igDbDir.toPath());
-
- String ig1DbWalPath = Paths.get(DFLT_STORE_DIR, "wal", igName).toString();
-
- U.delete(U.resolveWorkDirectory(U.defaultWorkDirectory(), ig1DbWalPath, false));
- }
-
- /** */
private void fillCache(
IgniteDataStreamer streamer,
int cacheSize,
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/file/IgnitePdsDiskErrorsRecoveringTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/file/IgnitePdsDiskErrorsRecoveringTest.java
index 42a8407..8af8903 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/file/IgnitePdsDiskErrorsRecoveringTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/file/IgnitePdsDiskErrorsRecoveringTest.java
@@ -48,6 +48,7 @@ import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccess
import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory;
import org.apache.ignite.internal.processors.cache.persistence.wal.FileWriteAheadLogManager;
import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Assert;
import org.junit.Test;
@@ -57,6 +58,7 @@ import static org.apache.ignite.IgniteSystemProperties.IGNITE_WAL_MMAP;
/**
* Tests node recovering after disk errors during interaction with persistent storage.
*/
+@WithSystemProperty(key = "IGNITE_PDS_CHECKPOINT_TEST_SKIP_SYNC", value = "false")
public class IgnitePdsDiskErrorsRecoveringTest extends GridCommonAbstractTest {
/** */
private static final int PAGE_SIZE = DataStorageConfiguration.DFLT_PAGE_SIZE;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxCrossCacheMapOnInvalidTopologyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxCrossCacheMapOnInvalidTopologyTest.java
new file mode 100644
index 0000000..73024f6
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxCrossCacheMapOnInvalidTopologyTest.java
@@ -0,0 +1,316 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.transactions;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.affinity.AffinityAssignment;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.GridCacheUtils;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
+
+/**
+ */
+public class TxCrossCacheMapOnInvalidTopologyTest extends GridCommonAbstractTest {
+ /** Partitions count. */
+ private static final int PARTS_CNT = 32;
+
+ /** Cache 1. */
+ private static final String CACHE1 = DEFAULT_CACHE_NAME;
+
+ /** Cache 2. */
+ private static final String CACHE2 = DEFAULT_CACHE_NAME + "2";
+
+ /** */
+ private static final int MB = 1024 * 1024;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setCommunicationSpi(new TestRecordingCommunicationSpi());
+ cfg.setClientMode("client".equals(igniteInstanceName));
+ cfg.setCacheConfiguration(cacheConfiguration(CACHE1), cacheConfiguration(CACHE2).setRebalanceOrder(10));
+
+ cfg.setDataStorageConfiguration(new DataStorageConfiguration().setPageSize(1024).
+ setDefaultDataRegionConfiguration(new DataRegionConfiguration().
+ setInitialSize(100 * MB).setMaxSize(100 * MB)));
+
+ return cfg;
+ }
+
+ /**
+ * @param name Name.
+ */
+ protected CacheConfiguration<Object, Object> cacheConfiguration(String name) {
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>(name);
+
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setBackups(2);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setOnheapCacheEnabled(false);
+ ccfg.setAffinity(new RendezvousAffinityFunction(false, PARTS_CNT));
+
+ return ccfg;
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testCrossCacheTxMapOnInvalidTopologyPessimistic() throws Exception {
+ doTestCrossCacheTxMapOnInvalidTopology(PESSIMISTIC, REPEATABLE_READ);
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testCrossCacheTxMapOnInvalidTopologyOptimistic() throws Exception {
+ doTestCrossCacheTxMapOnInvalidTopology(OPTIMISTIC, REPEATABLE_READ);
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testCrossCacheTxMapOnInvalidTopologyOptimisticSerializable() throws Exception {
+ doTestCrossCacheTxMapOnInvalidTopology(OPTIMISTIC, SERIALIZABLE);
+ }
+
+ /**
+ *
+ */
+ private void doTestCrossCacheTxMapOnInvalidTopology(TransactionConcurrency concurrency, TransactionIsolation isolation) throws Exception {
+ try {
+ IgniteEx crd = startGrid(0);
+ IgniteEx g1 = startGrid(1);
+
+ awaitPartitionMapExchange();
+
+ Ignite client = startGrid("client");
+ assertNotNull(client.cache(CACHE1));
+ assertNotNull(client.cache(CACHE2));
+
+ try (IgniteDataStreamer<Object, Object> streamer = crd.dataStreamer(CACHE1)) {
+ // Put 500 keys per partition.
+ for (int k = 0; k < PARTS_CNT * 500; k++)
+ streamer.addData(k, new byte[10]);
+ }
+
+ try (IgniteDataStreamer<Object, Object> streamer = crd.dataStreamer(CACHE2)) {
+ // Put 500 keys per partition.
+ for (int k = 0; k < PARTS_CNT * 500; k++)
+ streamer.addData(k, new byte[10]);
+ }
+
+ TestRecordingCommunicationSpi crdSpi = TestRecordingCommunicationSpi.spi(crd);
+
+ final AffinityTopologyVersion joinVer = new AffinityTopologyVersion(4, 0);
+ AffinityTopologyVersion leftVer = new AffinityTopologyVersion(5, 0);
+
+ AtomicReference<Set<Integer>> full = new AtomicReference<>();
+
+ crdSpi.blockMessages((node, m) -> {
+ if (m instanceof GridDhtPartitionSupplyMessage) {
+ GridDhtPartitionSupplyMessage msg = (GridDhtPartitionSupplyMessage)m;
+
+ // Allow full rebalance for cache 1 and system cache.
+ if (msg.groupId() == CU.cacheId(CACHE1) || msg.groupId() == CU.cacheId(GridCacheUtils.UTILITY_CACHE_NAME))
+ return false;
+
+ // Allow only first batch for cache 2.
+ if (msg.topologyVersion().equals(joinVer)) {
+ if (full.get() == null) {
+ Map<Integer, Long> last = U.field(msg, "last");
+
+ full.set(last.keySet());
+
+ return false;
+ }
+
+ return true;
+ }
+
+ if (msg.topologyVersion().equals(leftVer))
+ return true;
+ }
+
+ return false;
+ });
+
+ TestRecordingCommunicationSpi g1Spi = TestRecordingCommunicationSpi.spi(g1);
+ g1Spi.blockMessages((node, msg) -> {
+ if (msg instanceof GridDhtPartitionSupplyMessage) {
+ GridDhtPartitionSupplyMessage m = (GridDhtPartitionSupplyMessage)msg;
+
+ return m.groupId() == CU.cacheId(CACHE2);
+ }
+
+ return false;
+ });
+
+ startGrid(2);
+
+ crdSpi.waitForBlocked();
+ g1Spi.waitForBlocked();
+
+ // Wait partial owning.
+ assertTrue("Timed out while waiting for rebalance", GridTestUtils.waitForCondition(() -> {
+ // Await full rebalance for cache 2.
+ GridDhtPartitionTopology top0 = grid(2).cachex(CACHE1).context().topology();
+
+ for (int p = 0; p < PARTS_CNT; p++) {
+ if (top0.localPartition(p).state() != OWNING)
+ return false;
+ }
+
+ // Await partial rebalance for cache 1.
+ GridDhtPartitionTopology top1 = grid(2).cachex(CACHE2).context().topology();
+
+ for (Integer part : full.get()) {
+ if (top1.localPartition(part).state() != OWNING)
+ return false;
+ }
+
+ return true;
+ }, 10_000));
+
+ // At this point cache 1 is fully rebalanced and cache 2 is partially rebalanced.
+ // Stop supplier in the middle of rebalance.
+ g1.close();
+
+ // Wait for topologies and calculate required partitions.
+ grid(0).cachex(CACHE1).context().affinity().affinityReadyFuture(leftVer).get();
+ grid(2).cachex(CACHE1).context().affinity().affinityReadyFuture(leftVer).get();
+ grid(0).cachex(CACHE2).context().affinity().affinityReadyFuture(leftVer).get();
+ grid(2).cachex(CACHE2).context().affinity().affinityReadyFuture(leftVer).get();
+
+ AffinityAssignment assignment = grid(0).cachex(CACHE2).context().affinity().assignment(leftVer);
+
+ // Search for a partition with incompatible assignment.
+ int movingPart = -1;
+
+ for (int p = 0; p < assignment.assignment().size(); p++) {
+ List<ClusterNode> nodes = assignment.assignment().get(p);
+ List<ClusterNode> nodes2 = assignment.idealAssignment().get(p);
+
+ if (!nodes.equals(nodes2) && nodes.get(0).order() == 1)
+ movingPart = p;
+ }
+
+ assertFalse(movingPart == -1);
+
+ // Delay rebalance for next top ver.
+ TestRecordingCommunicationSpi.spi(grid(2)).blockMessages((node, message) -> {
+ if (message instanceof GridDhtPartitionsSingleMessage) {
+ GridDhtPartitionsSingleMessage sm = (GridDhtPartitionsSingleMessage)message;
+
+ return sm.exchangeId() != null;
+ }
+
+ return false;
+ });
+
+ TestRecordingCommunicationSpi.spi(client).blockMessages(new IgniteBiPredicate<ClusterNode, Message>() {
+ @Override public boolean apply(ClusterNode node, Message message) {
+ if (concurrency == PESSIMISTIC)
+ return message instanceof GridNearLockRequest;
+ else
+ return message instanceof GridNearTxPrepareRequest;
+ }
+ });
+
+ final int finalMovingPart = movingPart;
+ IgniteInternalFuture<?> txFut = multithreadedAsync(new Runnable() {
+ @Override public void run() {
+ try (Transaction tx = client.transactions().txStart(concurrency, isolation)) {
+ Map<Integer, Integer> map = new LinkedHashMap<>();
+
+ for (int p = 0; p < PARTS_CNT; p++)
+ map.put(p, p);
+
+ client.cache(CACHE1).putAll(map); // Will successfully lock topology.
+ client.cache(CACHE2).put(finalMovingPart, 0); // Should remap but will go through without fix.
+
+ tx.commit();
+ }
+ }
+ }, 1, "tx-thread");
+
+ TestRecordingCommunicationSpi.spi(client).waitForBlocked();
+
+ crdSpi.stopBlock(); // Continue rebalance and trigger next topology switch.
+
+ TestRecordingCommunicationSpi.spi(grid(2)).waitForBlocked();
+
+ TestRecordingCommunicationSpi.spi(client).stopBlock();
+
+ doSleep(5000); // Make sure request will listen for current topology future completion.
+
+ TestRecordingCommunicationSpi.spi(grid(2)).stopBlock();
+
+ crdSpi.stopBlock();
+
+ awaitPartitionMapExchange();
+
+ txFut.get();
+
+ checkFutures();
+ }
+ finally {
+ stopAllGrids();
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateAbstractTest.java
index fc53513..1d9a09d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateAbstractTest.java
@@ -78,6 +78,7 @@ import org.jetbrains.annotations.Nullable;
import static java.util.concurrent.TimeUnit.DAYS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.stream.Collectors.toCollection;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_BASELINE_AUTO_ADJUST_ENABLED;
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;
@@ -90,7 +91,7 @@ import static org.apache.ignite.testframework.GridTestUtils.runMultiThreadedAsyn
*/
public abstract class TxPartitionCounterStateAbstractTest extends GridCommonAbstractTest {
/** IP finder. */
- private static final TcpDiscoveryVmIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+ protected static final TcpDiscoveryVmIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
/** */
private static final int MB = 1024 * 1024;
@@ -143,10 +144,24 @@ public abstract class TxPartitionCounterStateAbstractTest extends GridCommonAbst
return cfg;
}
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ System.setProperty(IGNITE_BASELINE_AUTO_ADJUST_ENABLED, "false");
+
+ super.beforeTestsStarted();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ System.clearProperty(IGNITE_BASELINE_AUTO_ADJUST_ENABLED);
+ }
+
/**
* @param name Name.
*/
- protected CacheConfiguration cacheConfiguration(String name) {
+ protected CacheConfiguration<Object, Object> cacheConfiguration(String name) {
CacheConfiguration ccfg = new CacheConfiguration(name);
ccfg.setAtomicityMode(TRANSACTIONAL);
@@ -171,7 +186,7 @@ public abstract class TxPartitionCounterStateAbstractTest extends GridCommonAbst
stopAllGrids();
- //cleanPersistenceDir();
+ cleanPersistenceDir();
}
/**
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateConsistencyHistoryRebalanceTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateConsistencyHistoryRebalanceTest.java
index d1b3a1e..6cfa21b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateConsistencyHistoryRebalanceTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateConsistencyHistoryRebalanceTest.java
@@ -17,175 +17,13 @@
package org.apache.ignite.internal.processors.cache.transactions;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.IgniteEx;
-import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.TestRecordingCommunicationSpi;
-import org.apache.ignite.internal.processors.cache.CacheEntryInfoCollection;
-import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteBiPredicate;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.WithSystemProperty;
-import org.junit.Test;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD;
-import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
/**
* Test partitions consistency in various scenarios.
*/
@WithSystemProperty(key = IGNITE_PDS_WAL_REBALANCE_THRESHOLD, value = "0")
public class TxPartitionCounterStateConsistencyHistoryRebalanceTest extends TxPartitionCounterStateConsistencyTest {
- /**
- * Tests reproduces the problem: if partition is selected for full rebalancing for some reason and it was
- * rebalancing before it should always be cleared first OR deletes on supplier will not be visible on demander
- * causing partition desync.
- *
- * TODO FIXME https://issues.apache.org/jira/browse/IGNITE-11799
- *
- * @throws Exception If failed.
- */
- @Test
- public void testPartitionConsistencyDuringRebalanceAndConcurrentUpdates_MovingPartitionNotCleared() throws Exception {
- backups = 2;
-
- Ignite crd = startGridsMultiThreaded(SERVER_NODES);
-
- int[] primaryParts = crd.affinity(DEFAULT_CACHE_NAME).primaryPartitions(crd.cluster().localNode());
-
- IgniteCache<Object, Object> cache = crd.cache(DEFAULT_CACHE_NAME);
-
- List<Integer> p1Keys = partitionKeys(cache, primaryParts[0], 2, 0);
- List<Integer> p2Keys = partitionKeys(cache, primaryParts[1], 2, 0);
-
- cache.put(p1Keys.get(0), 0); // Will be historically rebalanced.
- cache.put(p1Keys.get(1), 1);
-
- forceCheckpoint();
-
- Ignite backup = backupNode(p1Keys.get(0), DEFAULT_CACHE_NAME);
-
- assertTrue(backupNodes(p2Keys.get(0), DEFAULT_CACHE_NAME).contains(backup));
-
- stopGrid(true, backup.name());
-
- cache.remove(p1Keys.get(1));
- cache.put(p2Keys.get(1), 1); // Will be fully rebalanced.
-
- TestRecordingCommunicationSpi spi = TestRecordingCommunicationSpi.spi(crd);
-
- // Prevent rebalance completion.
- spi.blockMessages(new IgniteBiPredicate<ClusterNode, Message>() {
- @Override public boolean apply(ClusterNode node, Message msg) {
- String name = (String)node.attributes().get(ATTR_IGNITE_INSTANCE_NAME);
-
- if (name.equals(backup.name()) && msg instanceof GridDhtPartitionSupplyMessage) {
- GridDhtPartitionSupplyMessage msg0 = (GridDhtPartitionSupplyMessage)msg;
-
- Map<Integer, CacheEntryInfoCollection> infos = U.field(msg0, "infos");
-
- return infos.keySet().contains(primaryParts[0]); // Delay historical rebalance.
- }
-
- return false;
- }
- });
-
- IgniteEx backup2 = startGrid(backup.name());
-
- doSleep(5000);
-
- spi.waitForBlocked();
- spi.stopBlock();
-
- // While message is delayed topology version shouldn't change to ideal.
- awaitPartitionMapExchange();
-
- assertPartitionsSame(idleVerify(crd, DEFAULT_CACHE_NAME));
- }
-
- /**
- * Same as {@link #testPartitionConsistencyDuringRebalanceAndConcurrentUpdates_MovingPartitionNotCleared} but
- * new coordinator is demander.
- *
- * @throws Exception If failed.
- */
- @Test
- public void testPartitionConsistencyCancelledRebalanceCoordinatorIsDemander() throws Exception {
- backups = 2;
-
- Ignite crd = startGrids(SERVER_NODES);
-
- crd.cluster().active(true);
-
- int[] primaryParts = crd.affinity(DEFAULT_CACHE_NAME).primaryPartitions(crd.cluster().localNode());
-
- IgniteCache<Object, Object> cache = crd.cache(DEFAULT_CACHE_NAME);
-
- List<Integer> p1Keys = partitionKeys(cache, primaryParts[0], 2, 0);
- List<Integer> p2Keys = partitionKeys(cache, primaryParts[1], 2, 0);
-
- assertTrue(crd.affinity(DEFAULT_CACHE_NAME).isPrimary(crd.cluster().localNode(), p1Keys.get(0)));
- assertTrue(crd.affinity(DEFAULT_CACHE_NAME).isPrimary(crd.cluster().localNode(), p2Keys.get(0)));
-
- final String primName = crd.name();
-
- cache.put(p1Keys.get(0), 0); // Will be historically rebalanced.
- cache.put(p1Keys.get(1), 1);
-
- forceCheckpoint();
-
- List<Ignite> backups = Arrays.asList(grid(1), grid(2));
-
- assertFalse(backups.contains(crd));
-
- final String demanderName = backups.get(0).name();
-
- stopGrid(true, demanderName);
-
- // Create counters delta.
- cache.remove(p1Keys.get(1));
- cache.put(p2Keys.get(1), 1); // Will be fully rebalanced.
-
- stopAllGrids();
-
- crd = startGrid(0);
- startGrid(1);
- startGrid(2);
-
- TestRecordingCommunicationSpi crdSpi = TestRecordingCommunicationSpi.spi(crd);
-
- // Block all rebalance from crd.
- crdSpi.blockMessages((node, msg) -> msg instanceof GridDhtPartitionSupplyMessage);
-
- crd.cluster().active(true);
-
- IgniteInternalFuture fut = GridTestUtils.runAsync(new Runnable() {
- @Override public void run() {
- try {
- crdSpi.waitForBlocked();
-
- // Stop before supplying rebalance. New rebalance must start with second backup as supplier
- // doing full rebalance.
- stopGrid(primName);
- }
- catch (InterruptedException e) {
- fail();
- }
- }
- });
-
- fut.get();
-
- awaitPartitionMapExchange();
-
- assertPartitionsSame(idleVerify(grid(demanderName), DEFAULT_CACHE_NAME));
- }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateConsistencyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateConsistencyTest.java
index c3f3664..183f90a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateConsistencyTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateConsistencyTest.java
@@ -17,19 +17,30 @@
package org.apache.ignite.internal.processors.cache.transactions;
+import java.lang.reflect.Field;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Random;
+import java.util.TreeMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Consumer;
+import java.util.stream.IntStream;
+
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteSystemProperties;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.cluster.ClusterTopologyException;
+import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
@@ -39,16 +50,20 @@ import org.apache.ignite.internal.pagemem.wal.WALPointer;
import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
import org.apache.ignite.internal.pagemem.wal.record.DataRecord;
import org.apache.ignite.internal.pagemem.wal.record.WALRecord;
+import org.apache.ignite.internal.processors.cache.CacheAffinityChangeMessage;
import org.apache.ignite.internal.processors.cache.CacheEntryInfoCollection;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
+import org.apache.ignite.internal.processors.cache.PartitionUpdateCounter;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage;
+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.util.typedef.G;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.lang.IgniteBiTuple;
-import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.discovery.tcp.BlockTcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.WithSystemProperty;
import org.apache.ignite.transactions.Transaction;
@@ -69,16 +84,30 @@ public class TxPartitionCounterStateConsistencyTest extends TxPartitionCounterSt
/** */
public static final int SERVER_NODES = 3;
+ /** */
+ protected TcpDiscoverySpi customDiscoSpi;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ if (customDiscoSpi != null) {
+ cfg.setDiscoverySpi(customDiscoSpi);
+
+ customDiscoSpi = null;
+ }
+
+ return cfg;
+ }
+
/**
- * Test if same updates order on all owners after txs are finished.
- *
- * @throws Exception If failed.
+ * Tests for same order of updates on all owners after txs are finished.
*/
@Test
public void testSingleThreadedUpdateOrder() throws Exception {
backups = 2;
- Ignite crd = startGridsMultiThreaded(SERVER_NODES);
+ startGridsMultiThreaded(SERVER_NODES);
IgniteEx client = startGrid("client");
@@ -185,7 +214,7 @@ public class TxPartitionCounterStateConsistencyTest extends TxPartitionCounterSt
assertFalse(backups.contains(prim));
- long stop = U.currentTimeMillis() + 3 * 60_000;
+ long stop = U.currentTimeMillis() + 60_000;
long seed = System.nanoTime();
@@ -197,7 +226,7 @@ public class TxPartitionCounterStateConsistencyTest extends TxPartitionCounterSt
IgniteInternalFuture<?> fut = multithreadedAsync(() -> {
while (U.currentTimeMillis() < stop) {
- doSleep(5_000);
+ doSleep(3_000);
Ignite restartNode = grid(1 + r.nextInt(backups.size()));
@@ -210,7 +239,7 @@ public class TxPartitionCounterStateConsistencyTest extends TxPartitionCounterSt
try {
waitForTopology(SERVER_NODES);
- doSleep(15_000);
+ doSleep(5000);
startGrid(name);
@@ -260,23 +289,21 @@ public class TxPartitionCounterStateConsistencyTest extends TxPartitionCounterSt
spi.blockMessages((node, msg) -> msg instanceof GridDhtPartitionSupplyMessage);
- IgniteInternalFuture fut = GridTestUtils.runAsync(new Runnable() {
- @Override public void run() {
- try {
- spi.waitForBlocked();
- }
- catch (InterruptedException e) {
- fail(X.getFullStackTrace(e));
- }
+ IgniteInternalFuture fut = GridTestUtils.runAsync(() -> {
+ try {
+ spi.waitForBlocked();
+ }
+ catch (InterruptedException e) {
+ fail(X.getFullStackTrace(e));
+ }
- prim.cache(DEFAULT_CACHE_NAME).remove(keys.get(0));
+ prim.cache(DEFAULT_CACHE_NAME).remove(keys.get(0));
- doSleep(2000);
+ doSleep(2000);
- prim.cache(DEFAULT_CACHE_NAME).remove(keys.get(1)); // Ensure queue cleanup is triggered.
+ prim.cache(DEFAULT_CACHE_NAME).remove(keys.get(1)); // Ensure queue cleanup is triggered.
- spi.stopBlock();
- }
+ spi.stopBlock();
});
startGrid(backups.get(0).name());
@@ -293,8 +320,6 @@ public class TxPartitionCounterStateConsistencyTest extends TxPartitionCounterSt
/**
* Tests reproduces the problem: in-place update in tree during rebalance in partition was not handled as update
* causing missed WAL record which has to be processed on recovery.
- *
- * @throws Exception
*/
@Test
public void testPartitionConsistencyDuringRebalanceAndConcurrentUpdates_CheckpointDuringRebalance() throws Exception {
@@ -328,20 +353,18 @@ public class TxPartitionCounterStateConsistencyTest extends TxPartitionCounterSt
TestRecordingCommunicationSpi spi = TestRecordingCommunicationSpi.spi(crd);
// Prevent rebalance completion.
- spi.blockMessages(new IgniteBiPredicate<ClusterNode, Message>() {
- @Override public boolean apply(ClusterNode node, Message msg) {
- String name = (String)node.attributes().get(ATTR_IGNITE_INSTANCE_NAME);
-
- if (name.equals(backupName) && msg instanceof GridDhtPartitionSupplyMessage) {
- GridDhtPartitionSupplyMessage msg0 = (GridDhtPartitionSupplyMessage)msg;
+ spi.blockMessages((node, msg) -> {
+ String name = (String)node.attributes().get(ATTR_IGNITE_INSTANCE_NAME);
- Map<Integer, CacheEntryInfoCollection> infos = U.field(msg0, "infos");
+ if (name.equals(backupName) && msg instanceof GridDhtPartitionSupplyMessage) {
+ GridDhtPartitionSupplyMessage msg0 = (GridDhtPartitionSupplyMessage)msg;
- return infos.keySet().contains(primaryParts[0]); // Delay historical rebalance.
- }
+ Map<Integer, CacheEntryInfoCollection> infos = U.field(msg0, "infos");
- return false;
+ return infos.keySet().contains(primaryParts[0]); // Delay historical rebalance.
}
+
+ return false;
});
backup = startGrid(backupName);
@@ -359,7 +382,7 @@ public class TxPartitionCounterStateConsistencyTest extends TxPartitionCounterSt
stopGrid(true, backupName);
- backup = startGrid(backupName);
+ startGrid(backupName);
awaitPartitionMapExchange();
@@ -367,6 +390,461 @@ public class TxPartitionCounterStateConsistencyTest extends TxPartitionCounterSt
}
/**
+ * Tests reproduces the problem: if coordinator is a demander after activation and supplier has left, new
+ * rebalance will finish and cause no partition inconsistencies.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testPartitionConsistencyCancelledRebalanceCoordinatorIsDemander() throws Exception {
+ backups = 2;
+
+ Ignite crd = startGrids(SERVER_NODES);
+
+ crd.cluster().active(true);
+
+ int[] primaryParts = crd.affinity(DEFAULT_CACHE_NAME).primaryPartitions(crd.cluster().localNode());
+
+ IgniteCache<Object, Object> cache = crd.cache(DEFAULT_CACHE_NAME);
+
+ List<Integer> p1Keys = partitionKeys(cache, primaryParts[0], 2, 0);
+
+ assertTrue(crd.affinity(DEFAULT_CACHE_NAME).isPrimary(crd.cluster().localNode(), p1Keys.get(0)));
+
+ final String primName = crd.name();
+
+ cache.put(p1Keys.get(0), 0);
+ cache.put(p1Keys.get(1), 1);
+
+ forceCheckpoint();
+
+ List<Ignite> backups = Arrays.asList(grid(1), grid(2));
+
+ assertFalse(backups.contains(crd));
+
+ final String demanderName = backups.get(0).name();
+
+ stopGrid(true, demanderName);
+
+ // Create counters delta.
+ cache.remove(p1Keys.get(1));
+
+ stopAllGrids();
+
+ crd = startGrid(0);
+ startGrid(1);
+ startGrid(2);
+
+ TestRecordingCommunicationSpi crdSpi = TestRecordingCommunicationSpi.spi(crd);
+
+ // Block all rebalance from crd.
+ crdSpi.blockMessages((node, msg) -> msg instanceof GridDhtPartitionSupplyMessage);
+
+ crd.cluster().active(true);
+
+ IgniteInternalFuture fut = GridTestUtils.runAsync(() -> {
+ try {
+ crdSpi.waitForBlocked();
+
+ // Stop before supplying rebalance. New rebalance must start with second backup as supplier
+ // doing full rebalance.
+ stopGrid(primName);
+ }
+ catch (InterruptedException e) {
+ fail();
+ }
+ });
+
+ fut.get();
+
+ awaitPartitionMapExchange();
+
+ assertPartitionsSame(idleVerify(grid(demanderName), DEFAULT_CACHE_NAME));
+ }
+
+ /**
+ * Tests reproduces the problem: if node joins after missing some updates no partition inconsistency happens.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testPartitionConsistencyDuringRebalanceAndConcurrentUpdates_NoOp() throws Exception {
+ testPartitionConsistencyDuringRebalanceConcurrentlyWithTopologyChange(s -> {
+ }, s -> {
+ });
+ }
+
+ /**
+ * Tests reproduces the problem: if node re-joins having MOVING partitions no partition inconsistency happens.
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testPartitionConsistencyDuringRebalanceConcurrentlyWithTopologyChange_DemanderRestart() throws Exception {
+ testPartitionConsistencyDuringRebalanceConcurrentlyWithTopologyChange(
+ demanderNodeName -> stopGrid(true, demanderNodeName),
+ demanderNodeName -> {
+ try {
+ startGrid(demanderNodeName);
+ }
+ catch (Exception e) {
+ fail(X.getFullStackTrace(e));
+ }
+ });
+ }
+
+ /**
+ * Tests reproduces the problem: if cache is started during rebalance no partition inconsistency happens.
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testPartitionConsistencyDuringRebalanceConcurrentlyWithTopologyChange_CacheStart() throws Exception {
+ testPartitionConsistencyDuringRebalanceConcurrentlyWithTopologyChange(
+ demanderNodeName -> grid(0).getOrCreateCache(cacheConfiguration(DEFAULT_CACHE_NAME + "2")),
+ demanderNodeName -> {});
+ }
+
+ /**
+ * Tests reproduces the problem: if non-BLT node is started during rebalance no partition inconsistency happens.
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testPartitionConsistencyDuringRebalanceConcurrentlyWithTopologyChange_NonBLTNodeStart() throws Exception {
+ testPartitionConsistencyDuringRebalanceConcurrentlyWithTopologyChange(
+ demanderNodeName -> {
+ try {
+ startGrid(SERVER_NODES);
+ }
+ catch (Exception e) {
+ fail(X.getFullStackTrace(e));
+ }
+ },
+ demanderNodeName -> {});
+ }
+
+ /** */
+ @Test
+ public void testPartitionConsistencyDuringRebalanceAndConcurrentUpdates_SameAffinityPME() throws Exception {
+ backups = 2;
+
+ Ignite crd = startGridsMultiThreaded(SERVER_NODES);
+
+ crd.cluster().active(true);
+
+ Ignite client = startGrid("client");
+
+ IgniteCache<Object, Object> cache = client.cache(DEFAULT_CACHE_NAME);
+
+ int threads = 8;
+
+ int keys = 200;
+ int batch = 10;
+
+ CyclicBarrier sync = new CyclicBarrier(threads + 1);
+
+ AtomicBoolean done = new AtomicBoolean();
+
+ Random r = new Random();
+
+ LongAdder puts = new LongAdder();
+ LongAdder restarts = new LongAdder();
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(() -> {
+ U.awaitQuiet(sync);
+
+ while(!done.get()) {
+ int batch0 = 1 + r.nextInt(batch - 1);
+ int start = r.nextInt(keys - batch0);
+
+ try(Transaction tx = client.transactions().txStart()) {
+ Map<Integer, Integer> map = new TreeMap<>();
+
+ IntStream.range(start, start + batch0).forEach(value -> map.put(value, value));
+
+ cache.putAll(map);
+
+ tx.commit();
+
+ puts.add(batch0);
+ }
+ }
+ }, threads, "load-thread");
+
+ IgniteInternalFuture fut2 = GridTestUtils.runAsync(() -> {
+ U.awaitQuiet(sync);
+
+ while(!done.get()) {
+ try {
+ IgniteCache cache1 = client.createCache(cacheConfiguration(DEFAULT_CACHE_NAME + "2"));
+
+ cache1.destroy();
+
+ restarts.increment();
+ }
+ catch (Exception e) {
+ fail(X.getFullStackTrace(e));
+ }
+ }
+ });
+
+ doSleep(30_000);
+
+ done.set(true);
+
+ fut.get();
+ fut2.get();
+
+ log.info("TX: puts=" + puts.sum() + ", restarts=" + restarts.sum() + ", size=" + cache.size());
+
+ assertPartitionsSame(idleVerify(client));
+ }
+
+ /**
+ * Tests tx load concurrently with PME not changing tx topology.
+ * In such scenario a race is possible with tx updates and PME counters set.
+ * Outdated counters on PME should be ignored.
+ */
+ @Test
+ public void testPartitionConsistencyDuringRebalanceAndConcurrentUpdates_TxDuringPME() throws Exception {
+ backups = 2;
+
+ Ignite crd = startGrid(0);
+ startGrid(1);
+ startGrid(2);
+
+ crd.cluster().active(true);
+
+ Ignite client = startGrid("client");
+
+ IgniteCache<Object, Object> cache = client.cache(DEFAULT_CACHE_NAME);
+
+ // Put one key per partition.
+ try(IgniteDataStreamer<Object, Object> streamer = client.dataStreamer(DEFAULT_CACHE_NAME)) {
+ for (int k = 0; k < PARTS_CNT; k++)
+ streamer.addData(k, 0);
+ }
+
+ Integer key0 = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME));
+ Integer key = primaryKey(grid(0).cache(DEFAULT_CACHE_NAME));
+
+ TestRecordingCommunicationSpi spi = TestRecordingCommunicationSpi.spi(crd);
+
+ spi.blockMessages((node, message) -> {
+ if (message instanceof GridDhtPartitionsFullMessage) {
+ GridDhtPartitionsFullMessage tmp = (GridDhtPartitionsFullMessage)message;
+
+ return tmp.exchangeId() != null;
+ }
+
+ return false;
+ });
+
+ // Locks mapped wait.
+ CountDownLatch l = new CountDownLatch(1);
+
+ IgniteInternalFuture fut = GridTestUtils.runAsync(() -> {
+ U.awaitQuiet(l);
+
+ try {
+ startGrid(SERVER_NODES);
+ }
+ catch (Exception e) {
+ fail(X.getFullStackTrace(e));
+ }
+ });
+
+ TestRecordingCommunicationSpi cliSpi = TestRecordingCommunicationSpi.spi(client);
+ cliSpi.blockMessages((node, message) -> {
+ // Block second lock map req.
+ return message instanceof GridNearLockRequest && node.order() == crd.cluster().localNode().order();
+ });
+
+ IgniteInternalFuture txFut = GridTestUtils.runAsync(() -> {
+ try(Transaction tx = client.transactions().txStart()) {
+ Map<Integer, Integer> map = new LinkedHashMap<>();
+
+ map.put(key, key); // clientFirst=true in lockAll.
+ map.put(key0, key0); // clientFirst=false in lockAll.
+
+ cache.putAll(map);
+
+ tx.commit(); // Will start preparing in the middle of PME.
+ }
+ });
+
+ IgniteInternalFuture lockFut = GridTestUtils.runAsync(() -> {
+ try {
+ cliSpi.waitForBlocked(); // Delay first before PME.
+
+ l.countDown();
+
+ spi.waitForBlocked(); // Block PME after finish on crd and wait on others.
+
+ cliSpi.stopBlock(); // Start remote lock mapping.
+ }
+ catch (InterruptedException e) {
+ fail();
+ }
+ });
+
+ txFut.get(); // Transaction should not be blocked by concurrent PME.
+ lockFut.get();
+
+ spi.stopBlock();
+
+ fut.get();
+
+ awaitPartitionMapExchange();
+
+ assertPartitionsSame(idleVerify(crd, DEFAULT_CACHE_NAME));
+
+ // Expect correct reservation counters.
+ PartitionUpdateCounter cntr = counter(key, grid(0).name());
+ assertNotNull(cntr);
+ assertEquals(cntr.toString(), 2, cntr.reserved());
+ }
+
+ /**
+ * Tests tx load concurrently with PME for switching late affinity.
+ * <p>
+ * Scenario: two keys tx mapped locally on late affinity topology and when mapped and prepared remotely on ideal
+ * topology, first key is mapped to non-moving partition, second is mapped on moving partition.
+ * <p>
+ * Success: key over moving partition is prepared on new owner (choosed after late affinity switch),
+ * otherwise it's possible txs are prepared on different primaries after late affinity switch.
+ */
+ @Test
+ public void testPartitionConsistencyDuringRebalanceAndConcurrentUpdates_LateAffinitySwitch() throws Exception {
+ backups = 1;
+
+ customDiscoSpi = new BlockTcpDiscoverySpi().setIpFinder(IP_FINDER);
+
+ Field rndAddrsField = U.findField(BlockTcpDiscoverySpi.class, "skipAddrsRandomization");
+ assertNotNull(rndAddrsField);
+ rndAddrsField.set(customDiscoSpi, true);
+
+ Ignite crd = startGrid(0); // Start coordinator with custom discovery SPI.
+ IgniteEx g1 = startGrid(1);
+ startGrid(2);
+
+ crd.cluster().active(true);
+
+ // Same name pattern as in test configuration.
+ String consistentId = "node" + getTestIgniteInstanceName(3);
+
+ List<Integer> g1Keys = primaryKeys(g1.cache(DEFAULT_CACHE_NAME), 10);
+ List<Integer> movingFromG1 = movingKeysAfterJoin(g1, DEFAULT_CACHE_NAME, 10, null, consistentId);
+
+ // Retain only stable keys;
+ g1Keys.removeAll(movingFromG1);
+
+ // The key will move from grid0 to grid3.
+ Integer key = movingKeysAfterJoin(crd, DEFAULT_CACHE_NAME, 1, null, consistentId).get(0);
+
+ IgniteEx g3 = startGrid(3);
+
+ assertEquals(consistentId, g3.localNode().consistentId());
+
+ resetBaselineTopology();
+ awaitPartitionMapExchange();
+
+ assertTrue(crd.affinity(DEFAULT_CACHE_NAME).isPrimary(g1.localNode(), g1Keys.get(0)));
+
+ stopGrid(3);
+
+ Ignite client = startGrid("client");
+
+ IgniteCache<Object, Object> cache = client.cache(DEFAULT_CACHE_NAME);
+ IgniteCache<Object, Object> cache2 = client.getOrCreateCache(cacheConfiguration(DEFAULT_CACHE_NAME + "2"));
+
+ // Put one key per partition.
+ for (int k = 0; k < PARTS_CNT; k++) {
+ cache.put(k, 0);
+ cache2.put(k, 0);
+ }
+
+ CountDownLatch resumeDiscoSndLatch = new CountDownLatch(1);
+
+ BlockTcpDiscoverySpi crdDiscoSpi = (BlockTcpDiscoverySpi)grid(0).configuration().getDiscoverySpi();
+ CyclicBarrier sync = new CyclicBarrier(2);
+
+ crdDiscoSpi.setClosure((node, msg) -> {
+ if (msg instanceof CacheAffinityChangeMessage) {
+ U.awaitQuiet(sync);
+ U.awaitQuiet(resumeDiscoSndLatch);
+ }
+
+ return null;
+ });
+
+ // Locks mapped wait.
+ IgniteInternalFuture fut = GridTestUtils.runAsync(() -> {
+ try {
+ startGrid(SERVER_NODES);
+
+ awaitPartitionMapExchange();
+ }
+ catch (Exception e) {
+ fail(X.getFullStackTrace(e));
+ }
+ });
+
+ sync.await();
+
+ TestRecordingCommunicationSpi clientSpi = TestRecordingCommunicationSpi.spi(client);
+ clientSpi.blockMessages((node, msg) -> msg instanceof GridNearLockRequest);
+
+ IgniteInternalFuture txFut = GridTestUtils.runAsync(() -> {
+ try (Transaction tx = client.transactions().txStart()) {
+ Map<Integer, Integer> map = new LinkedHashMap<>();
+
+ map.put(g1Keys.get(0), g1Keys.get(0)); // clientFirst=true in lockAll mapped to stable part.
+ map.put(key, key); // clientFirst=false in lockAll mapped to moving part.
+
+ cache.putAll(map);
+ cache2.putAll(new LinkedHashMap<>(map));
+
+ tx.commit(); // Will start preparing in the middle of PME.
+ }
+ });
+
+ IgniteInternalFuture lockFut = GridTestUtils.runAsync(() -> {
+ try {
+ // Wait for first lock request sent on local (late) topology.
+ clientSpi.waitForBlocked();
+ // Continue late switch PME.
+ resumeDiscoSndLatch.countDown();
+ crdDiscoSpi.setClosure(null);
+
+ // Wait late affinity switch.
+ awaitPartitionMapExchange();
+ // Continue tx mapping and preparing.
+ clientSpi.stopBlock();
+ }
+ catch (InterruptedException e) {
+ fail(X.getFullStackTrace(e));
+ }
+ });
+
+ fut.get();
+ txFut.get();
+ lockFut.get();
+
+ assertPartitionsSame(idleVerify(crd, DEFAULT_CACHE_NAME));
+
+ // TX must be prepared over new owner.
+ PartitionUpdateCounter cntr = counter(key, grid(3).name());
+ assertNotNull(cntr);
+ assertEquals(cntr.toString(), 2, cntr.reserved());
+
+ PartitionUpdateCounter cntr2 = counter(key, DEFAULT_CACHE_NAME + "2", grid(3).name());
+ assertNotNull(cntr2);
+ assertEquals(cntr2.toString(), 2, cntr2.reserved());
+ }
+
+ /**
* @param ignite Ignite.
*/
private WALIterator walIterator(IgniteEx ignite) throws IgniteCheckedException {
@@ -467,6 +945,71 @@ public class TxPartitionCounterStateConsistencyTest extends TxPartitionCounterSt
}, Runtime.getRuntime().availableProcessors() * 2, "tx-update-thread");
}
+ /**
+ * @param rebBlockClo Closure called after supply message is blocked in the middle of rebalance.
+ * @param rebUnblockClo Closure called after supply message is unblocked.
+ *
+ * @throws Exception If failed.
+ */
+ protected void testPartitionConsistencyDuringRebalanceConcurrentlyWithTopologyChange(
+ Consumer<String> rebBlockClo,
+ Consumer<String> rebUnblockClo)
+ throws Exception {
+ backups = 2;
+
+ Ignite crd = startGridsMultiThreaded(SERVER_NODES);
+
+ int[] primaryParts = crd.affinity(DEFAULT_CACHE_NAME).primaryPartitions(crd.cluster().localNode());
+
+ IgniteCache<Object, Object> cache = crd.cache(DEFAULT_CACHE_NAME);
+
+ List<Integer> keys = partitionKeys(cache, primaryParts[0], 2, 0);
+
+ cache.put(keys.get(0), 0);
+ cache.put(keys.get(1), 0);
+
+ forceCheckpoint();
+
+ Ignite backup = backupNode(keys.get(0), DEFAULT_CACHE_NAME);
+
+ final String backupName = backup.name();
+
+ stopGrid(false, backupName);
+
+ cache.remove(keys.get(1));
+
+ TestRecordingCommunicationSpi spi = TestRecordingCommunicationSpi.spi(crd);
+
+ // Prevent rebalance completion.
+ spi.blockMessages((node, msg) -> {
+ String name = (String)node.attributes().get(ATTR_IGNITE_INSTANCE_NAME);
+
+ if (name.equals(backupName) && msg instanceof GridDhtPartitionSupplyMessage) {
+ GridDhtPartitionSupplyMessage msg0 = (GridDhtPartitionSupplyMessage)msg;
+
+ Map<Integer, CacheEntryInfoCollection> infos = U.field(msg0, "infos");
+
+ return infos.keySet().contains(primaryParts[0]);
+ }
+
+ return false;
+ });
+
+ startGrid(backupName);
+
+ spi.waitForBlocked();
+
+ rebBlockClo.accept(backupName);
+
+ spi.stopBlock();
+
+ rebUnblockClo.accept(backupName);
+
+ awaitPartitionMapExchange();
+
+ assertPartitionsSame(idleVerify(crd, DEFAULT_CACHE_NAME));
+ }
+
/** */
private static class TestVal {
/** */
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnMapOnInvalidTopologyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnMapOnInvalidTopologyTest.java
new file mode 100644
index 0000000..b8bc99d
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnMapOnInvalidTopologyTest.java
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.transactions;
+
+import java.util.List;
+import java.util.concurrent.locks.ReadWriteLock;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopologyImpl;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionRollbackException;
+import org.junit.Test;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.EVICTED;
+
+/**
+ * Tests an ability to rollback transactions on invalid mapping during topology change.
+ */
+public class TxRollbackOnMapOnInvalidTopologyTest extends GridCommonAbstractTest {
+ /** */
+ private static final int GRIDS = 3;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setConsistentId(igniteInstanceName);
+
+ cfg.setCommunicationSpi(new TestRecordingCommunicationSpi());
+
+ cfg.setClientMode(igniteInstanceName.equals("client"));
+
+ CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME);
+
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setBackups(2);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ startGridsMultiThreaded(GRIDS);
+
+ startGrid("client");
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /**
+ * Tests rollbacks when mapped to invalid topology.
+ */
+ @Test
+ public void testRollbackOnMapToInvalidTopology_1() throws Exception {
+ doTestRollback(grid("client"), grid(0));
+ }
+
+ /**
+ * Tests rollbacks when mapped to invalid topology.
+ */
+ @Test
+ public void testRollbackOnMapToInvalidTopology_2() throws Exception {
+ doTestRollback(grid("client"), grid(1));
+ }
+
+ /**
+ * Tests rollbacks when mapped to invalid topology.
+ */
+ @Test
+ public void testRollbackOnMapToInvalidTopology_3() throws Exception {
+ doTestRollback(grid("client"), grid(2));
+ }
+
+ /**
+ * Tests rollbacks when mapped to invalid topology.
+ */
+ @Test
+ public void testRollbackOnMapToInvalidTopology_4() throws Exception {
+ doTestRollback(grid(0), grid(0));
+ }
+
+ /**
+ * Tests rollbacks when mapped to invalid topology.
+ */
+ @Test
+ public void testRollbackOnMapToInvalidTopology_5() throws Exception {
+ doTestRollback(grid(0), grid(1));
+ }
+
+ /**
+ * Tests rollbacks when mapped to invalid topology.
+ */
+ @Test
+ public void testRollbackOnMapToInvalidTopology_6() throws Exception {
+ doTestRollback(grid(0), grid(2));
+ }
+
+ /**
+ * Test scenario: mock partition to fail check, start new node.
+ * Expected result: Transaction is rolled back.
+ *
+ * @param near Near mode.
+ * @param node Owner.
+ */
+ private void doTestRollback(Ignite near, IgniteEx node) throws Exception {
+ List<Integer> primKeys = primaryKeys(node.cache(DEFAULT_CACHE_NAME), 100);
+ List<Integer> movingKeys = movingKeysAfterJoin(node, DEFAULT_CACHE_NAME, 100);
+
+ primKeys.removeAll(movingKeys); /** {@code primKeys} contains stable partitions. */
+
+ int part = primKeys.get(0);
+
+ IgniteEx grid = (IgniteEx)grid(node.affinity(DEFAULT_CACHE_NAME).mapPartitionToNode(part));
+
+ GridDhtPartitionTopologyImpl top =
+ (GridDhtPartitionTopologyImpl)grid.cachex(DEFAULT_CACHE_NAME).context().topology();
+
+ AffinityTopologyVersion failCheckVer = new AffinityTopologyVersion(GRIDS + 2, 1);
+
+ top.partitionFactory((ctx, grp, id) -> new GridDhtLocalPartition(ctx, grp, id, false) {
+ @Override public boolean primary(AffinityTopologyVersion topVer) {
+ return !(id == part && topVer.equals(failCheckVer)) && super.primary(topVer);
+ }
+ });
+
+ // Re-create mocked part.
+ GridDhtLocalPartition p0 = top.localPartition(part);
+ p0.rent(false).get();
+ assertTrue(p0.state() == EVICTED);
+
+ ReadWriteLock lock = U.field(top, "lock");
+ lock.writeLock().lock();
+ p0 = top.getOrCreatePartition(part);
+ p0.own();
+ lock.writeLock().unlock();
+
+ startGrid(GRIDS);
+ awaitPartitionMapExchange();
+
+ try(Transaction tx = near.transactions().txStart()) {
+ near.cache(DEFAULT_CACHE_NAME).put(part, part);
+
+ tx.commit();
+
+ fail();
+ }
+ catch (TransactionRollbackException ignore) {
+ // Expected.
+ }
+ catch (Exception e) {
+ fail(X.getFullStackTrace(e));
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/BlockTcpDiscoverySpi.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/BlockTcpDiscoverySpi.java
new file mode 100644
index 0000000..23b88a6
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/BlockTcpDiscoverySpi.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.spi.discovery.tcp;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.Socket;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.managers.discovery.CustomMessageWrapper;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteBiClosure;
+import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCustomEventMessage;
+
+import static org.junit.Assert.assertNotNull;
+
+/**
+ * Custom discovery SPI allowing to block custom messages transfer between nodes.
+ */
+public class BlockTcpDiscoverySpi extends TcpDiscoverySpi {
+ /** Closure. */
+ private volatile IgniteBiClosure<ClusterNode, DiscoveryCustomMessage, Void> clo;
+
+ /**
+ * @param clo Closure.
+ */
+ public void setClosure(IgniteBiClosure<ClusterNode, DiscoveryCustomMessage, Void> clo) {
+ this.clo = clo;
+ }
+
+ /**
+ * @param addr Address.
+ * @param msg Message.
+ */
+ private synchronized void apply(ClusterNode addr, TcpDiscoveryAbstractMessage msg) {
+ if (!(msg instanceof TcpDiscoveryCustomEventMessage))
+ return;
+
+ TcpDiscoveryCustomEventMessage cm = (TcpDiscoveryCustomEventMessage)msg;
+
+ DiscoveryCustomMessage delegate;
+
+ try {
+ DiscoverySpiCustomMessage custMsg = cm.message(marshaller(), U.resolveClassLoader(ignite().configuration()));
+
+ assertNotNull(custMsg);
+
+ delegate = ((CustomMessageWrapper)custMsg).delegate();
+
+ }
+ catch (Throwable throwable) {
+ throw new RuntimeException(throwable);
+ }
+
+ if (clo != null)
+ clo.apply(addr, delegate);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void writeToSocket(
+ Socket sock,
+ TcpDiscoveryAbstractMessage msg,
+ byte[] data,
+ long timeout
+ ) throws IOException {
+ if (spiCtx != null)
+ apply(spiCtx.localNode(), msg);
+
+ super.writeToSocket(sock, msg, data, timeout);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void writeToSocket(Socket sock,
+ OutputStream out,
+ TcpDiscoveryAbstractMessage msg,
+ long timeout) throws IOException, IgniteCheckedException {
+ if (spiCtx != null)
+ apply(spiCtx.localNode(), msg);
+
+ super.writeToSocket(sock, out, msg, timeout);
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryReconnectUnstableTopologyTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryReconnectUnstableTopologyTest.java
index c69280a..8428f41 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryReconnectUnstableTopologyTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryReconnectUnstableTopologyTest.java
@@ -17,29 +17,18 @@
package org.apache.ignite.spi.discovery.tcp;
-import java.io.IOException;
-import java.io.OutputStream;
import java.lang.reflect.Field;
-import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.DiscoverySpiTestListener;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.managers.discovery.CustomMessageWrapper;
-import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteBiClosure;
-import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
-import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
-import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCustomEventMessage;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
@@ -150,71 +139,4 @@ public class TcpDiscoveryReconnectUnstableTopologyTest extends GridCommonAbstrac
private TcpDiscoverySpi spi(Ignite ig) {
return (TcpDiscoverySpi)ig.configuration().getDiscoverySpi();
}
-
- /**
- * Discovery SPI with blocking support.
- */
- protected class BlockTcpDiscoverySpi extends TcpDiscoverySpi {
- /** Closure. */
- private volatile IgniteBiClosure<ClusterNode, DiscoveryCustomMessage, Void> clo;
-
- /**
- * @param clo Closure.
- */
- public void setClosure(IgniteBiClosure<ClusterNode, DiscoveryCustomMessage, Void> clo) {
- this.clo = clo;
- }
-
- /**
- * @param addr Address.
- * @param msg Message.
- */
- private synchronized void apply(ClusterNode addr, TcpDiscoveryAbstractMessage msg) {
- if (!(msg instanceof TcpDiscoveryCustomEventMessage))
- return;
-
- TcpDiscoveryCustomEventMessage cm = (TcpDiscoveryCustomEventMessage)msg;
-
- DiscoveryCustomMessage delegate;
-
- try {
- DiscoverySpiCustomMessage custMsg = cm.message(marshaller(), U.resolveClassLoader(ignite().configuration()));
-
- assertNotNull(custMsg);
-
- delegate = ((CustomMessageWrapper)custMsg).delegate();
-
- }
- catch (Throwable throwable) {
- throw new RuntimeException(throwable);
- }
-
- if (clo != null)
- clo.apply(addr, delegate);
- }
-
- /** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
- TcpDiscoveryAbstractMessage msg,
- byte[] data,
- long timeout
- ) throws IOException {
- if (spiCtx != null)
- apply(spiCtx.localNode(), msg);
-
- super.writeToSocket(sock, msg, data, timeout);
- }
-
- /** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock,
- OutputStream out,
- TcpDiscoveryAbstractMessage msg,
- long timeout) throws IOException, IgniteCheckedException {
- if (spiCtx != null)
- apply(spiCtx.localNode(), msg);
-
- super.writeToSocket(sock, out, msg, timeout);
- }
- }
}
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 1630c16..f612156 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
@@ -102,6 +102,7 @@ import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.PA;
import org.apache.ignite.internal.util.typedef.T2;
+import org.apache.ignite.internal.util.typedef.T3;
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.visor.VisorTaskArgument;
@@ -124,6 +125,7 @@ import org.apache.ignite.transactions.TransactionRollbackException;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import static java.util.stream.Collectors.toList;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED;
import static org.apache.ignite.IgniteSystemProperties.getBoolean;
import static org.apache.ignite.cache.CacheMode.LOCAL;
@@ -858,6 +860,8 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
sb.append("nodeId=")
.append(k.context().localNodeId())
+ .append(" consistentId=")
+ .append(k.localNode().consistentId())
.append(" isDone=")
.append(syncFut.isDone())
.append("\n");
@@ -935,10 +939,11 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
if (part != null) {
sb.append(p).append(" counters=")
- .append(part == null ? "NA" : part.dataStore().partUpdateCounter())
+ .append(part.dataStore().partUpdateCounter())
.append(" fullSize=")
- .append(part == null ? "NA" : part.fullSize())
- .append(" state=").append(part.state());
+ .append(part.fullSize())
+ .append(" state=").append(part.state())
+ .append(" reservations=").append(part.reservations());
}
else
sb.append(p).append(" is null");
@@ -958,7 +963,7 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
.append(" fullSize=")
.append(part == null ? "NA" : part.fullSize())
.append(" state=")
- .append(part == null ? "NA" : top.partitionState(nodeId, p))
+ .append(top.partitionState(nodeId, p))
.append(" isAffNode=")
.append(affNodes.contains(nodeId))
.append("\n");
@@ -1291,7 +1296,7 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
* @return List of keys.
*/
protected final List<Integer> movingKeysAfterJoin(Ignite ign, String cacheName, int size) {
- return movingKeysAfterJoin(ign, cacheName, size, null);
+ return movingKeysAfterJoin(ign, cacheName, size, null, null);
}
/**
@@ -1302,11 +1307,13 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
* @param cacheName Cache name.
* @param size Number of keys.
* @param nodeInitializer Node initializer closure.
+ * @param joiningNodeConsistentId Joining node consistent id.
* @return List of keys.
*/
protected final List<Integer> movingKeysAfterJoin(Ignite ign, String cacheName, int size,
- @Nullable IgniteInClosure<ClusterNode> nodeInitializer) {
- assertEquals("Expected consistentId is set to node name", ign.name(), ign.cluster().localNode().consistentId());
+ @Nullable IgniteInClosure<ClusterNode> nodeInitializer, @Nullable String joiningNodeConsistentId) {
+ if (joiningNodeConsistentId == null)
+ assertEquals("Expected consistentId is set to node name", ign.name(), ign.cluster().localNode().consistentId());
ArrayList<ClusterNode> nodes = new ArrayList<>(ign.cluster().nodes());
@@ -1317,7 +1324,8 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
if (nodeInitializer != null)
nodeInitializer.apply(fakeNode);
- fakeNode.consistentId(getTestIgniteInstanceName(nodes.size()));
+ fakeNode.consistentId(joiningNodeConsistentId == null ? getTestIgniteInstanceName(nodes.size()) :
+ joiningNodeConsistentId);
nodes.add(fakeNode);
@@ -2312,20 +2320,26 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
protected void assertCountersSame(int partId, boolean withReserveCntr) throws AssertionFailedError {
PartitionUpdateCounter cntr0 = null;
- for (Ignite ignite : G.allGrids()) {
- if (ignite.configuration().isClientMode())
- continue;
+ List<T3<String, @Nullable PartitionUpdateCounter, Boolean>> cntrMap = G.allGrids().stream().filter(ignite ->
+ !ignite.configuration().isClientMode()).map(ignite ->
+ new T3<>(ignite.name(), counter(partId, ignite.name()),
+ ignite.affinity(DEFAULT_CACHE_NAME).isPrimary(ignite.cluster().localNode(), partId))).collect(toList());
- PartitionUpdateCounter cntr = counter(partId, ignite.name());
+ for (T3<String, PartitionUpdateCounter, Boolean> cntr : cntrMap) {
+ if (cntr.get2() == null)
+ continue;
if (cntr0 != null) {
- assertEquals("Expecting same counters [partId=" + partId + ']', cntr0, cntr);
+ assertEquals("Expecting same counters [partId=" + partId +
+ ", cntrs=" + cntrMap + ']', cntr0, cntr.get2());
if (withReserveCntr)
- assertEquals("Expecting same reservation counters", cntr0.reserved(), cntr.reserved());
+ assertEquals("Expecting same reservation counters [partId=" + partId +
+ ", cntrs=" + cntrMap + ']',
+ cntr0.reserved(), cntr.get2().reserved());
}
- cntr0 = cntr;
+ cntr0 = cntr.get2();
}
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite7.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite7.java
index 86e4d41..b9c6365 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite7.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite7.java
@@ -35,6 +35,7 @@ import org.apache.ignite.internal.processors.cache.distributed.rebalancing.GridC
import org.apache.ignite.internal.processors.cache.distributed.rebalancing.GridCacheRebalancingWithAsyncClearingMvccTest;
import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionMultinodeMixedRegionsTest;
import org.apache.ignite.internal.processors.cache.persistence.db.CheckpointBufferDeadlockTest;
+import org.apache.ignite.internal.processors.cache.transactions.TxCrossCacheMapOnInvalidTopologyTest;
import org.apache.ignite.testframework.junits.DynamicSuite;
import org.junit.runner.RunWith;
@@ -68,6 +69,9 @@ public class IgniteCacheMvccTestSuite7 {
ignoredTests.add(CacheGroupMetricsTest.class); // See MvccCacheGroupMetricsMBeanTest
ignoredTests.add(GridCacheRebalancingPartitionCountersTest.class); // See GridCacheRebalancingPartitionCountersMvccTest
+ // Test logic is not compatible with MVCC style tx locking.
+ ignoredTests.add(TxCrossCacheMapOnInvalidTopologyTest.class);
+
List<Class<?>> suite = new ArrayList<>(IgniteCacheTestSuite7.suite(ignoredTests));
// Add Mvcc clones.
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite6.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite6.java
index ac462f6..ee02b56 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite6.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite6.java
@@ -51,6 +51,7 @@ import org.apache.ignite.internal.processors.cache.transactions.TxOptimisticPrep
import org.apache.ignite.internal.processors.cache.transactions.TxRollbackAsyncNearCacheTest;
import org.apache.ignite.internal.processors.cache.transactions.TxRollbackAsyncTest;
import org.apache.ignite.internal.processors.cache.transactions.TxRollbackOnIncorrectParamsTest;
+import org.apache.ignite.internal.processors.cache.transactions.TxRollbackOnMapOnInvalidTopologyTest;
import org.apache.ignite.internal.processors.cache.transactions.TxRollbackOnTimeoutNearCacheTest;
import org.apache.ignite.internal.processors.cache.transactions.TxRollbackOnTimeoutNoDeadlockDetectionTest;
import org.apache.ignite.internal.processors.cache.transactions.TxRollbackOnTimeoutOnePhaseCommitTest;
@@ -141,6 +142,8 @@ public class IgniteCacheTestSuite6 {
//GridTestUtils.addTestIfNeeded(suite, GridCacheRebalancingOrderingTest.class, ignoredTests);
//GridTestUtils.addTestIfNeeded(suite, IgniteCacheClientMultiNodeUpdateTopologyLockTest.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite, TxRollbackOnMapOnInvalidTopologyTest.class, ignoredTests);
+
return suite;
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite7.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite7.java
index a940203..c3fb301 100755
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite7.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite7.java
@@ -37,6 +37,7 @@ import org.apache.ignite.internal.processors.cache.WalModeChangeCoordinatorNotAf
import org.apache.ignite.internal.processors.cache.WalModeChangeSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.Cache64kPartitionsTest;
import org.apache.ignite.internal.processors.cache.distributed.CacheDataLossOnPartitionMoveTest;
+import org.apache.ignite.internal.processors.cache.distributed.CachePartitionLostWhileClearingTest;
import org.apache.ignite.internal.processors.cache.distributed.CacheRentingStateRepairTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.IgniteCacheStartWithLoadTest;
import org.apache.ignite.internal.processors.cache.distributed.rebalancing.GridCacheRebalancingPartitionCountersTest;
@@ -45,6 +46,7 @@ import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionMu
import org.apache.ignite.internal.processors.cache.persistence.IgnitePdsCacheAssignmentNodeRestartsTest;
import org.apache.ignite.internal.processors.cache.persistence.db.CheckpointBufferDeadlockTest;
import org.apache.ignite.internal.processors.cache.transactions.TransactionIntegrityWithPrimaryIndexCorruptionTest;
+import org.apache.ignite.internal.processors.cache.transactions.TxCrossCacheMapOnInvalidTopologyTest;
import org.apache.ignite.internal.processors.cache.transactions.TxRollbackAsyncWithPersistenceTest;
import org.apache.ignite.internal.processors.cache.transactions.TxWithSmallTimeoutAndContentionOneKeyTest;
import org.apache.ignite.testframework.GridTestUtils;
@@ -109,6 +111,10 @@ public class IgniteCacheTestSuite7 {
GridTestUtils.addTestIfNeeded(suite, CacheConfigurationSerializationOnDiscoveryTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite, CacheConfigurationSerializationOnExchangeTest.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite, CachePartitionLostWhileClearingTest.class, ignoredTests);
+
+ GridTestUtils.addTestIfNeeded(suite, TxCrossCacheMapOnInvalidTopologyTest.class, ignoredTests);
+
return suite;
}
}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/BinaryMetadataConcurrentUpdateWithIndexesTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/BinaryMetadataConcurrentUpdateWithIndexesTest.java
index 955b2f0..dc74fe5 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/BinaryMetadataConcurrentUpdateWithIndexesTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/BinaryMetadataConcurrentUpdateWithIndexesTest.java
@@ -17,10 +17,7 @@
package org.apache.ignite.internal.processors.cache;
-import java.io.IOException;
-import java.io.OutputStream;
import java.lang.reflect.Field;
-import java.net.Socket;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -28,7 +25,6 @@ import java.util.LinkedHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.binary.BinaryObjectBuilder;
import org.apache.ignite.binary.BinaryObjectException;
@@ -39,7 +35,6 @@ import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.QueryIndex;
import org.apache.ignite.cache.QueryIndexType;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
@@ -48,17 +43,11 @@ import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.binary.BinaryMetadata;
-import org.apache.ignite.internal.managers.discovery.CustomMessageWrapper;
-import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl;
import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateProposedMessage;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteBiClosure;
-import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
-import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
-import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
-import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCustomEventMessage;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.spi.discovery.tcp.BlockTcpDiscoverySpi;
import org.apache.ignite.transactions.Transaction;
import org.junit.Test;
@@ -351,73 +340,6 @@ public class BinaryMetadataConcurrentUpdateWithIndexesTest extends GridCommonAbs
return builder.build();
}
- /**
- * Discovery SPI which can simulate network split.
- */
- protected class BlockTcpDiscoverySpi extends TcpDiscoverySpi {
- /** Closure. */
- private volatile IgniteBiClosure<ClusterNode, DiscoveryCustomMessage, Void> clo;
-
- /**
- * @param clo Closure.
- */
- public void setClosure(IgniteBiClosure<ClusterNode, DiscoveryCustomMessage, Void> clo) {
- this.clo = clo;
- }
-
- /**
- * @param addr Address.
- * @param msg Message.
- */
- private synchronized void apply(ClusterNode addr, TcpDiscoveryAbstractMessage msg) {
- if (!(msg instanceof TcpDiscoveryCustomEventMessage))
- return;
-
- TcpDiscoveryCustomEventMessage cm = (TcpDiscoveryCustomEventMessage)msg;
-
- DiscoveryCustomMessage delegate;
-
- try {
- DiscoverySpiCustomMessage custMsg = cm.message(marshaller(), U.resolveClassLoader(ignite().configuration()));
-
- assertNotNull(custMsg);
-
- delegate = ((CustomMessageWrapper)custMsg).delegate();
-
- }
- catch (Throwable throwable) {
- throw new RuntimeException(throwable);
- }
-
- if (clo != null)
- clo.apply(addr, delegate);
- }
-
- /** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
- TcpDiscoveryAbstractMessage msg,
- byte[] data,
- long timeout
- ) throws IOException {
- if (spiCtx != null)
- apply(spiCtx.localNode(), msg);
-
- super.writeToSocket(sock, msg, data, timeout);
- }
-
- /** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock,
- OutputStream out,
- TcpDiscoveryAbstractMessage msg,
- long timeout) throws IOException, IgniteCheckedException {
- if (spiCtx != null)
- apply(spiCtx.localNode(), msg);
-
- super.writeToSocket(sock, out, msg, timeout);
- }
- }
-
/** {@inheritDoc} */
@Override protected void beforeTest() throws Exception {
super.beforeTest();
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccTxRecoveryTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccTxRecoveryTest.java
index c3cd0e3..d5a2e4f 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccTxRecoveryTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccTxRecoveryTest.java
@@ -43,10 +43,12 @@ import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManager;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareRequest;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFinishResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxManager;
import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyImpl;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
@@ -57,6 +59,7 @@ import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.transactions.Transaction;
import org.junit.Test;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
import static org.apache.ignite.internal.processors.cache.mvcc.CacheMvccTxRecoveryTest.NodeMode.CLIENT;
@@ -508,6 +511,81 @@ public class CacheMvccTxRecoveryTest extends CacheMvccAbstractTest {
* @throws Exception if failed.
*/
@Test
+ public void testTxRecoveryWithLostFullMessageOnJoiningBackupNode() throws Exception {
+ CountDownLatch success = new CountDownLatch(1);
+
+ int joiningBackupNodeId = 2;
+
+ IgniteEx crd = startGrid(0);
+
+ IgniteEx partOwner = startGrid(1);
+
+ IgniteCache<Object, Object> cache = partOwner.getOrCreateCache(new CacheConfiguration<>(DEFAULT_CACHE_NAME)
+ .setAtomicityMode(TRANSACTIONAL)
+ .setCacheMode(PARTITIONED)
+ .setIndexedTypes(Integer.class, Integer.class)
+ .setBackups(2));
+
+ // prevent FullMassage on joining backup node
+ ((TestRecordingCommunicationSpi)crd.configuration().getCommunicationSpi())
+ .blockMessages(GridDhtPartitionsFullMessage.class, getTestIgniteInstanceName(joiningBackupNodeId));
+
+ new Thread(() -> {
+ try {
+ startGrid(joiningBackupNodeId);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ success.countDown();
+ }).start();
+
+ assertTrue(GridTestUtils.waitForCondition(() -> crd.cluster().nodes().size() == 3, 10_000));
+
+ ArrayList<Integer> keys = new ArrayList<>();
+
+ Affinity<Object> aff = crd.affinity(DEFAULT_CACHE_NAME);
+
+ for (int i = 0; i < 100; i++) {
+ if (aff.isPrimary(partOwner.localNode(), i)) {
+ keys.add(i);
+ break;
+ }
+ }
+
+ ((TestRecordingCommunicationSpi)partOwner.configuration().getCommunicationSpi())
+ .blockMessages(GridDhtTxPrepareRequest.class, getTestIgniteInstanceName(joiningBackupNodeId));
+
+ GridNearTxLocal nearTx = ((TransactionProxyImpl)partOwner.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)).tx();
+
+ for (Integer k : keys)
+ cache.put(k, k);
+
+ nearTx.commitAsync();
+
+ // Checks that PartitionCountersNeighborcastRequest will be sent after primary node left.
+ IgniteTxManager tm = crd.context().cache().context().tm();
+ assertTrue(GridTestUtils.waitForCondition(() -> !tm.activeTransactions().isEmpty(), 10_000));
+ assertTrue(GridTestUtils.waitForCondition(() -> tm.activeTransactions().iterator().next().state().equals(PREPARED), 10_000));
+
+ // Primary node left.
+ partOwner.close();
+
+ // Node with backup fetch lost FullMessage and starts.
+ ((TestRecordingCommunicationSpi)crd.configuration().getCommunicationSpi()).stopBlock();
+
+ success.await();
+
+ awaitPartitionMapExchange();
+
+ assertEquals(2, crd.cluster().nodes().size());
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ @Test
public void testUpdateCountersGapIsClosed() throws Exception {
int srvCnt = 3;