You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by mp...@apache.org on 2022/11/22 13:16:22 UTC
[ignite] branch master updated: IGNITE-17737 Fixed snapshot inconsistency if it was taken under cache workload. (#10345)
This is an automated email from the ASF dual-hosted git repository.
mpetrov 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 491a963cc76 IGNITE-17737 Fixed snapshot inconsistency if it was taken under cache workload. (#10345)
491a963cc76 is described below
commit 491a963cc76312c54085a66e076281e4beb3cdbd
Author: Mikhail Petrov <32...@users.noreply.github.com>
AuthorDate: Tue Nov 22 16:16:12 2022 +0300
IGNITE-17737 Fixed snapshot inconsistency if it was taken under cache workload. (#10345)
---
.../internal/processors/cache/ExchangeContext.java | 22 ++
.../processors/cache/GridCacheAffinityManager.java | 35 +++
.../cache/distributed/dht/GridDhtCacheAdapter.java | 30 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 2 +-
.../distributed/dht/atomic/GridDhtAtomicCache.java | 2 +-
.../preloader/GridDhtPartitionsExchangeFuture.java | 4 +
.../snapshot/IgniteSnapshotManager.java | 2 +-
.../cache/transactions/IgniteTxHandler.java | 35 +--
.../snapshot/AbstractSnapshotSelfTest.java | 2 +-
.../snapshot/IgniteSnapshotConsistencyTest.java | 309 +++++++++++++++++++++
.../ignite/testsuites/IgniteSnapshotTestSuite.java | 4 +-
11 files changed, 384 insertions(+), 63 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
index c7eb5a019e9..4e2197e319d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
@@ -22,6 +22,7 @@ import java.util.Set;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.SystemProperty;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -54,6 +55,17 @@ public class ExchangeContext {
/** PME is not required. */
private boolean exchangeFreeSwitch;
+ /**
+ * Whether cache requests that were mapped to an outdated topology version should be forcibly remapped to
+ * the up-to-date one. {@code True} forces all cache requests (including those that were mapped to the stale topology
+ * version with the same key affinity distribution) to wait current PME completion on all nodes.
+ * This is critical in situations when you want to ensure that no cache requests can be executed until PME is
+ * finished on all nodes (e.g. snapshot operation).
+ *
+ * @see GridCacheAffinityManager#isCompatibleWithCurrentTopologyVersion(AffinityTopologyVersion)
+ */
+ private final boolean remapStaleCacheReq;
+
/** Merges allowed flag. */
private final boolean merge;
@@ -103,6 +115,8 @@ public class ExchangeContext {
}
evts = new ExchangeDiscoveryEvents(fut);
+
+ remapStaleCacheReq = isSnapshotOperation(fut.firstEvent());
}
/**
@@ -159,6 +173,14 @@ public class ExchangeContext {
return merge;
}
+ /**
+ * @return Whether cache requests that were mapped to an outdated topology version should be forcibly remapped to
+ * the up-to-date one.
+ */
+ public boolean remapStaleCacheRequests() {
+ return remapStaleCacheReq;
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(ExchangeContext.class, this);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
index 8b551040998..8c659133525 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
@@ -402,4 +402,39 @@ public class GridCacheAffinityManager extends GridCacheManagerAdapter {
return aff0.primaryChanged(part, startVer, endVer);
}
+
+ /**
+ * @param rmtVer Topology version to check.
+ * @return Whether the specified topology version can be considered identical to the current topology version.
+ */
+ public boolean isCompatibleWithCurrentTopologyVersion(AffinityTopologyVersion rmtVer) {
+ AffinityTopologyVersion curVer = cctx.topology().readyTopologyVersion();
+
+ if (curVer.equals(rmtVer))
+ return true;
+
+ AffinityTopologyVersion lastAffChangedTopVer = cctx.shared().exchange().lastAffinityChangedTopologyVersion(rmtVer);
+
+ if (curVer.isBetween(lastAffChangedTopVer, rmtVer))
+ return true;
+
+ if (cctx.shared().exchange().lastFinishedFuture().context().remapStaleCacheRequests())
+ return false;
+
+ Collection<ClusterNode> cacheNodes0 = cctx.discovery().cacheGroupAffinityNodes(cctx.groupId(), rmtVer);
+ Collection<ClusterNode> cacheNodes1 = cctx.discovery().cacheGroupAffinityNodes(cctx.groupId(), curVer);
+
+ if (!cacheNodes0.equals(cacheNodes1) || affinityTopologyVersion().before(curVer))
+ return false;
+
+ try {
+ List<List<ClusterNode>> aff1 = assignments(rmtVer);
+ List<List<ClusterNode>> aff2 = assignments(curVer);
+
+ return aff1.equals(aff2);
+ }
+ catch (IllegalStateException ignored) {
+ return false;
+ }
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
index 90b8ec8ebce..557153c0495 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
@@ -1203,36 +1203,12 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
}
/**
- * @param mapVer Mapped topology version.
- * @param curVer Current topology version.
+ * @param rmtVer Topology version that the cache request was mapped to by the remote node.
* @return {@code True} if cache affinity changed and operation should be remapped.
*/
- protected final boolean needRemap(AffinityTopologyVersion mapVer, AffinityTopologyVersion curVer) {
- if (curVer.equals(mapVer))
- return false;
-
- AffinityTopologyVersion lastAffChangedTopVer = ctx.shared().exchange().lastAffinityChangedTopologyVersion(mapVer);
-
- if (curVer.isBetween(lastAffChangedTopVer, mapVer))
- return false;
-
+ protected final boolean needRemap(AffinityTopologyVersion rmtVer) {
// TODO IGNITE-7164 check mvcc crd for mvcc enabled txs.
-
- Collection<ClusterNode> cacheNodes0 = ctx.discovery().cacheGroupAffinityNodes(ctx.groupId(), mapVer);
- Collection<ClusterNode> cacheNodes1 = ctx.discovery().cacheGroupAffinityNodes(ctx.groupId(), curVer);
-
- if (!cacheNodes0.equals(cacheNodes1) || ctx.affinity().affinityTopologyVersion().before(curVer))
- return true;
-
- try {
- List<List<ClusterNode>> aff1 = ctx.affinity().assignments(mapVer);
- List<List<ClusterNode>> aff2 = ctx.affinity().assignments(curVer);
-
- return !aff1.equals(aff2);
- }
- catch (IllegalStateException ignored) {
- return true;
- }
+ return !ctx.affinity().isCompatibleWithCurrentTopologyVersion(rmtVer);
}
/**
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 9e7ff00924f..fcd7ab1d3be 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
@@ -1085,7 +1085,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
}
try {
- if (top != null && needRemap(req.topologyVersion(), top.readyTopologyVersion())) {
+ if (top != null && needRemap(req.topologyVersion())) {
if (log.isDebugEnabled()) {
log.debug("Client topology version mismatch, need remap lock request [" +
"reqTopVer=" + req.topologyVersion() +
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 07d76d71747..6b00da0ecfa 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -1826,7 +1826,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
// Can not wait for topology future since it will break
// GridNearAtomicCheckUpdateRequest processing.
remap = !compatible && !top.topologyVersionFuture().isDone() ||
- needRemap(req.topologyVersion(), top.readyTopologyVersion());
+ needRemap(req.topologyVersion());
}
if (!remap) {
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 cd4d77c918c..faf80aa04a9 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
@@ -152,6 +152,10 @@ import static org.apache.ignite.internal.util.IgniteUtils.doInParallelUninterrup
/**
* Future for exchanging partition maps.
+ * <p>
+ * Note, that by default cache requests mapped to a stale topology version can be executed even if PME is in still in
+ * progress on back-up nodes but is finished on primary node. If this behaviour is unacceptable see
+ * {@link ExchangeContext#remapStaleCacheRequests()} flag for more details.
*/
@SuppressWarnings({"TypeMayBeWeakened", "unchecked"})
public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapter
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
index 75867df52fb..04cddffc847 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
@@ -1715,7 +1715,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
/** {@inheritDoc} */
@Override public void onDoneBeforeTopologyUnlock(GridDhtPartitionsExchangeFuture fut) {
- if (clusterSnpReq == null || cctx.kernalContext().clientNode())
+ if (clusterSnpReq == null || cctx.kernalContext().clientNode() || !isSnapshotOperation(fut.firstEvent()))
return;
SnapshotOperationRequest snpReq = clusterSnpReq;
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 77bf58fa869..4e9d37007ce 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
@@ -459,7 +459,7 @@ public class IgniteTxHandler {
}
}
- if (!retry && needRemap(req.topologyVersion(), top.readyTopologyVersion(), req)) {
+ if (!retry && needRemap(req.topologyVersion(), req)) {
retry = true;
if (txPrepareMsgLog.isDebugEnabled()) {
@@ -708,43 +708,16 @@ public class IgniteTxHandler {
}
/**
- * @param expVer Expected topology version.
- * @param curVer Current topology version.
+ * @param rmtVer Topology version that the cache request was mapped to by the remote node.
* @param req Request.
* @return {@code True} if cache affinity changed and request should be remapped.
*/
- private boolean needRemap(AffinityTopologyVersion expVer,
- AffinityTopologyVersion curVer,
- GridNearTxPrepareRequest req) {
- if (curVer.equals(expVer))
- return false;
-
- AffinityTopologyVersion lastAffChangedTopVer = ctx.exchange().lastAffinityChangedTopologyVersion(expVer);
-
- if (curVer.compareTo(expVer) <= 0 && curVer.compareTo(lastAffChangedTopVer) >= 0)
- return false;
-
+ private boolean needRemap(AffinityTopologyVersion rmtVer, GridNearTxPrepareRequest req) {
// TODO IGNITE-6754 check mvcc crd for mvcc enabled txs.
for (IgniteTxEntry e : F.concat(false, req.reads(), req.writes())) {
- GridCacheContext ctx = e.context();
-
- Collection<ClusterNode> cacheNodes0 = ctx.discovery().cacheGroupAffinityNodes(ctx.groupId(), expVer);
- Collection<ClusterNode> cacheNodes1 = ctx.discovery().cacheGroupAffinityNodes(ctx.groupId(), curVer);
-
- if (!cacheNodes0.equals(cacheNodes1) || ctx.affinity().affinityTopologyVersion().compareTo(curVer) < 0)
+ if (!e.context().affinity().isCompatibleWithCurrentTopologyVersion(rmtVer))
return true;
-
- try {
- List<List<ClusterNode>> aff1 = ctx.affinity().assignments(expVer);
- List<List<ClusterNode>> aff2 = ctx.affinity().assignments(curVer);
-
- if (!aff1.equals(aff2))
- return true;
- }
- catch (IllegalStateException ignored) {
- return true;
- }
}
return false;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java
index 9aa661a6acb..702c8a9986b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java
@@ -113,7 +113,7 @@ import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
@RunWith(Parameterized.class)
public abstract class AbstractSnapshotSelfTest extends GridCommonAbstractTest {
/** Default snapshot name. */
- protected static final String SNAPSHOT_NAME = "testSnapshot";
+ public static final String SNAPSHOT_NAME = "testSnapshot";
/** Number of cache keys to pre-create at node start. */
protected static final int CACHE_KEYS_RANGE = 1024;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotConsistencyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotConsistencyTest.java
new file mode 100644
index 00000000000..75ca2969053
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotConsistencyTest.java
@@ -0,0 +1,309 @@
+/*
+ * 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.persistence.snapshot;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheAtomicityMode;
+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.managers.communication.GridMessageListener;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.distributed.GridCacheModuloAffinityFunction;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicSingleUpdateRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointListener;
+import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
+import org.apache.ignite.internal.util.future.IgniteFutureImpl;
+import org.apache.ignite.lang.IgniteFuture;
+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.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static java.util.Collections.singletonMap;
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cluster.ClusterState.ACTIVE;
+import static org.apache.ignite.internal.GridTopic.TOPIC_CACHE;
+import static org.apache.ignite.internal.TestRecordingCommunicationSpi.spi;
+import static org.apache.ignite.internal.processors.cache.distributed.GridCacheModuloAffinityFunction.IDX_ATTR;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.AbstractSnapshotSelfTest.SNAPSHOT_NAME;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/** Tests the consistency of taken snapshots across multiple nodes. */
+@RunWith(Parameterized.class)
+public class IgniteSnapshotConsistencyTest extends GridCommonAbstractTest {
+ /** */
+ private final AtomicInteger keyCntr = new AtomicInteger(0);
+
+ /** */
+ private final Map<Integer, BlockingCheckpointListener> blockedCheckpointNodes = new ConcurrentHashMap<>();
+
+ /** */
+ @Parameterized.Parameter
+ public boolean isOpInitiatorClient;
+
+ /** */
+ @Parameterized.Parameter(1)
+ public CacheAtomicityMode atomicity;
+
+ /** */
+ @Parameterized.Parameter(2)
+ public int backups;
+
+ /** */
+ @Parameterized.Parameter(3)
+ public TransactionConcurrency txConcurrency;
+
+ /** */
+ @Parameterized.Parameters(name = "isClient={0}, atomicity={1}, backups={2}, txConcurrency={3}")
+ public static Iterable<Object[]> data() {
+ List<Object[]> res = new ArrayList<>();
+
+ for (boolean isClient : Arrays.asList(true, false)) {
+ for (int backups = 1; backups <= 2; ++backups) {
+ res.add(new Object[] {isClient, ATOMIC, backups, null});
+
+ for (TransactionConcurrency txConcurrency : TransactionConcurrency.values())
+ res.add(new Object[] {isClient, TRANSACTIONAL, backups, txConcurrency});
+ }
+ }
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setDataStorageConfiguration(new DataStorageConfiguration()
+ .setDefaultDataRegionConfiguration(new DataRegionConfiguration()
+ .setPersistenceEnabled(true)));
+
+ cfg.setConsistentId(igniteInstanceName);
+ cfg.setUserAttributes(singletonMap(IDX_ATTR, getTestIgniteInstanceIndex(igniteInstanceName)));
+ cfg.setCommunicationSpi(new TestRecordingCommunicationSpi());
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ cleanPersistenceDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+
+ cleanPersistenceDir();
+ }
+
+ /**
+ * Tests the following scenario:
+ * 1. Cache put is performed while snapshot creation operation is in progress.
+ * 2. Cache put is mapped to a topology version that is earlier than the version associated with the start of the
+ * snapshot operation.
+ * 3. Check that snapshots taken from all nodes are consistent.
+ */
+ @Test
+ public void testConcurrentPutWithStaleTopologyVersion() throws Exception {
+ int srvCnt = 3;
+ int primaryIdx = 1;
+ int hangingBackupIdx = 2;
+
+ IgniteEx crd = startGrids(srvCnt);
+ IgniteEx opInitiator = isOpInitiatorClient ? startClientGrid(srvCnt) : startGrid(srvCnt);
+
+ crd.cluster().state(ACTIVE);
+
+ crd.getOrCreateCache(new CacheConfiguration<>(DEFAULT_CACHE_NAME)
+ .setBackups(backups)
+ .setAffinity(new GridCacheModuloAffinityFunction(srvCnt, backups))
+ .setAtomicityMode(atomicity));
+
+ awaitPartitionMapExchange();
+
+ for (int i = 0; i < srvCnt; i++)
+ grid(i).cache(DEFAULT_CACHE_NAME).put(keyForNode(i), 0);
+
+ forceCheckpoint();
+
+ spi(opInitiator).blockMessages(
+ atomicity == TRANSACTIONAL ? GridNearTxPrepareRequest.class : GridNearAtomicSingleUpdateRequest.class,
+ grid(primaryIdx).name()
+ );
+
+ IgniteInternalFuture<Void> putFut = doPut(opInitiator, keyForNode(primaryIdx));
+
+ spi(opInitiator).waitForBlocked(1, getTestTimeout());
+
+ blockCheckpoint(hangingBackupIdx);
+
+ AffinityTopologyVersion snpTopVer = grid(1).context().cache().context()
+ .exchange()
+ .lastTopologyFuture()
+ .topologyVersion()
+ .nextMinorVersion();
+
+ IgniteFuture<Void> snpFut = crd.snapshot().createSnapshot(SNAPSHOT_NAME);
+
+ waitForReadyTopology(grid(primaryIdx).cachex(DEFAULT_CACHE_NAME).context().topology(), snpTopVer);
+
+ AtomicBoolean isPutRemappedOnNewTop = new AtomicBoolean(false);
+
+ addNewTopologyVersionRemapListener(primaryIdx, /*initiator node idx*/ srvCnt, snpTopVer, isPutRemappedOnNewTop);
+
+ spi(opInitiator).stopBlock();
+
+ GridTestUtils.waitForCondition(() -> isPutRemappedOnNewTop.get() || putFut.isDone(), getTestTimeout());
+
+ unblockCheckpoint(hangingBackupIdx);
+
+ snpFut.get(getTestTimeout());
+ putFut.get(getTestTimeout());
+
+ IdleVerifyResultV2 snpVerifyRes = crd.context().cache().context().snapshotMgr().checkSnapshot(SNAPSHOT_NAME, null).get();
+
+ assertFalse(snpVerifyRes.hasConflicts());
+ }
+
+ /** */
+ private void addNewTopologyVersionRemapListener(
+ int locNodeIdx,
+ int rmtNodeIdx,
+ AffinityTopologyVersion newTopVer,
+ AtomicBoolean remapFlag
+ ) {
+ grid(locNodeIdx).context().io().addMessageListener(TOPIC_CACHE, new GridMessageListener() {
+ @Override public void onMessage(UUID nodeId, Object msg, byte plc) {
+ if (
+ (msg instanceof GridNearAtomicSingleUpdateRequest || msg instanceof GridNearTxPrepareRequest)
+ && nodeId.equals(grid(rmtNodeIdx).localNode().id())
+ && ((GridCacheMessage)msg).topologyVersion().compareTo(newTopVer) == 0
+ )
+ remapFlag.set(true);
+ }
+ });
+ }
+
+ /** */
+ private IgniteInternalFuture<Void> doPut(Ignite ignite, int key) {
+ IgniteCache<Integer, Object> cache = ignite.cache(DEFAULT_CACHE_NAME);
+
+ IgniteInternalFuture<Void> fut;
+
+ if (atomicity == TRANSACTIONAL) {
+ fut = GridTestUtils.runAsync(() -> {
+ try (Transaction tx = ignite.transactions().txStart(txConcurrency, REPEATABLE_READ)) {
+ cache.put(key, "val");
+
+ tx.commit();
+ }
+ });
+ }
+ else
+ fut = ((IgniteFutureImpl<Void>)cache.putAsync(key, "val")).internalFuture();
+
+ return fut;
+ }
+
+ /** */
+ private void blockCheckpoint(int nodeIdx) {
+ BlockingCheckpointListener blockLsnr = new BlockingCheckpointListener();
+
+ ((GridCacheDatabaseSharedManager)grid(nodeIdx).context().cache().context().database())
+ .addCheckpointListener(blockLsnr);
+
+ blockedCheckpointNodes.put(nodeIdx, blockLsnr);
+ }
+
+ /** */
+ private void unblockCheckpoint(int nodeIdx) {
+ BlockingCheckpointListener lsnr = blockedCheckpointNodes.remove(nodeIdx);
+
+ if (lsnr == null)
+ return;
+
+ lsnr.unblock();
+
+ ((GridCacheDatabaseSharedManager)grid(nodeIdx).context().cache().context().database())
+ .removeCheckpointListener(lsnr);
+ }
+
+ /** */
+ private int keyForNode(int nodeIdx) {
+ return keyForNode(grid(0).affinity(DEFAULT_CACHE_NAME), keyCntr, grid(nodeIdx).cluster().localNode());
+ }
+
+ /** */
+ private static class BlockingCheckpointListener implements CheckpointListener {
+ /** */
+ private final CountDownLatch checkpointUnblockedLatch = new CountDownLatch(1);
+
+ /** {@inheritDoc} */
+ @Override public void onMarkCheckpointBegin(CheckpointListener.Context ctx) throws IgniteCheckedException {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onCheckpointBegin(CheckpointListener.Context ctx) throws IgniteCheckedException {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void beforeCheckpointBegin(CheckpointListener.Context ctx) throws IgniteCheckedException {
+ try {
+ checkpointUnblockedLatch.await();
+ }
+ catch (InterruptedException e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ /** */
+ public void unblock() {
+ checkpointUnblockedLatch.countDown();
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite.java
index fccc3574389..f151ba19739 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite.java
@@ -24,6 +24,7 @@ import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteCl
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteClusterSnapshotSelfTest;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteClusterSnapshotStreamerTest;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteClusterSnapshotWalRecordTest;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotConsistencyTest;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotMXBeanTest;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManagerSelfTest;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotRemoteRequestTest;
@@ -48,7 +49,8 @@ import org.junit.runners.Suite;
PlainSnapshotTest.class,
EncryptedSnapshotTest.class,
IgniteClusterSnapshotWalRecordTest.class,
- IgniteClusterSnapshotStreamerTest.class
+ IgniteClusterSnapshotStreamerTest.class,
+ IgniteSnapshotConsistencyTest.class
})
public class IgniteSnapshotTestSuite {
}