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 6c41ed2 IGNITE-12328 IgniteException "Failed to resolve nodes
topology" during cache.removeAll() and constantly changing topology - Fixes
#7015.
6c41ed2 is described below
commit 6c41ed2ac91070d334a2c4eddf8a4dfebc44037b
Author: Alexey Scherbakov <alexey.scherbak...@gmail.com>
AuthorDate: Tue Oct 29 14:22:00 2019 +0300
IGNITE-12328 IgniteException "Failed to resolve nodes topology" during
cache.removeAll() and constantly changing topology - Fixes #7015.
Signed-off-by: Ivan Rakov <ira...@apache.org>
---
.../processors/cache/GridCacheIdMessage.java | 2 +-
.../cache/distributed/dht/GridDhtCacheAdapter.java | 17 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 2 +-
.../distributed/dht/atomic/GridDhtAtomicCache.java | 10 +-
.../dht/colocated/GridDhtColocatedLockFuture.java | 39 +-
.../cache/distributed/near/GridNearLockFuture.java | 4 -
.../cache/ClientSlowDiscoveryAbstractTest.java | 121 ++++++
... => ClientSlowDiscoveryTopologyChangeTest.java} | 105 +----
.../ClientSlowDiscoveryTransactionRemapTest.java | 473 +++++++++++++++++++++
.../TxCrossCacheMapOnInvalidTopologyTest.java | 133 ++++--
.../TxPartitionCounterStateConsistencyTest.java | 24 +-
.../testsuites/IgniteCacheMvccTestSuite5.java | 2 +
.../ignite/testsuites/IgniteCacheTestSuite5.java | 6 +-
13 files changed, 758 insertions(+), 180 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIdMessage.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIdMessage.java
index e094439..84d890c 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIdMessage.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIdMessage.java
@@ -112,6 +112,6 @@ public abstract class GridCacheIdMessage extends
GridCacheMessage {
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(GridCacheIdMessage.class, this);
+ return S.toString(GridCacheIdMessage.class, this, "super",
super.toString());
}
}
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 31adc06..da03d00 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
@@ -1274,30 +1274,29 @@ public abstract class GridDhtCacheAdapter<K, V> extends
GridDistributedCacheAdap
}
/**
- * @param expVer Expected topology version.
+ * @param mapVer Mapped topology version.
* @param curVer Current topology version.
* @return {@code True} if cache affinity changed and operation should be
remapped.
*/
- protected final boolean needRemap(AffinityTopologyVersion expVer,
AffinityTopologyVersion curVer,
- Collection<KeyCacheObject> keys) {
- if (curVer.equals(expVer))
+ protected final boolean needRemap(AffinityTopologyVersion mapVer,
AffinityTopologyVersion curVer) {
+ if (curVer.equals(mapVer))
return false;
- AffinityTopologyVersion lastAffChangedTopVer =
ctx.shared().exchange().lastAffinityChangedTopologyVersion(expVer);
+ AffinityTopologyVersion lastAffChangedTopVer =
ctx.shared().exchange().lastAffinityChangedTopologyVersion(mapVer);
- if (curVer.compareTo(lastAffChangedTopVer) >= 0 &&
curVer.compareTo(expVer) <= 0)
+ if (curVer.isBetween(lastAffChangedTopVer, mapVer))
return false;
// TODO IGNITE-7164 check mvcc crd for mvcc enabled txs.
- Collection<ClusterNode> cacheNodes0 =
ctx.discovery().cacheGroupAffinityNodes(ctx.groupId(), expVer);
+ 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().compareTo(curVer) < 0)
+ if (!cacheNodes0.equals(cacheNodes1) ||
ctx.affinity().affinityTopologyVersion().before(curVer))
return true;
try {
- List<List<ClusterNode>> aff1 = ctx.affinity().assignments(expVer);
+ List<List<ClusterNode>> aff1 = ctx.affinity().assignments(mapVer);
List<List<ClusterNode>> aff2 = ctx.affinity().assignments(curVer);
return !aff1.equals(aff2);
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 8fdc6fd..c7f30c7 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
@@ -1083,7 +1083,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
}
try {
- if (top != null && needRemap(req.topologyVersion(),
top.readyTopologyVersion(), req.keys())) {
+ if (top != null && needRemap(req.topologyVersion(),
top.readyTopologyVersion())) {
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 4c47c0c..926530d 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
@@ -1836,10 +1836,16 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
// Do not check topology version if topology was
locked on near node by
// external transaction or explicit lock.
if (!req.topologyLocked()) {
+ AffinityTopologyVersion waitVer =
top.topologyVersionFuture().initialVersion();
+
+ // No need to remap if next future version is
compatible.
+ boolean compatible =
+
waitVer.isBetween(req.lastAffinityChangedTopologyVersion(),
req.topologyVersion());
+
// Can not wait for topology future since it
will break
// GridNearAtomicCheckUpdateRequest processing.
- remap = !top.topologyVersionFuture().isDone()
||
- needRemap(req.topologyVersion(),
top.readyTopologyVersion(), req.keys());
+ remap = !compatible &&
!top.topologyVersionFuture().isDone() ||
+ needRemap(req.topologyVersion(),
top.readyTopologyVersion());
}
if (!remap) {
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 6731405..b0a0e6d 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
@@ -178,6 +178,9 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
/** {@code True} when mappings are ready for processing. */
private boolean mappingsReady;
+ /** */
+ private boolean trackable = true;
+
/**
* @param cctx Registry.
* @param keys Keys to lock.
@@ -262,12 +265,12 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
/** {@inheritDoc} */
@Override public boolean trackable() {
- return true;
+ return trackable;
}
/** {@inheritDoc} */
@Override public void markNotTrackable() {
- throw new UnsupportedOperationException();
+ trackable = false;
}
/**
@@ -772,20 +775,24 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
topVer = tx.topologyVersionSnapshot();
if (topVer != null) {
- for (GridDhtTopologyFuture fut :
cctx.shared().exchange().exchangeFutures()) {
- if (fut.exchangeDone() &&
fut.topologyVersion().equals(topVer)) {
- Throwable err = null;
+ AffinityTopologyVersion lastChangeVer =
cctx.shared().exchange().lastAffinityChangedTopologyVersion(topVer);
- // Before cache validation, make sure that this topology
future is already completed.
- try {
- fut.get();
- }
- catch (IgniteCheckedException e) {
- err = fut.error();
- }
+ IgniteInternalFuture<AffinityTopologyVersion> affFut =
cctx.shared().exchange().affinityReadyFuture(lastChangeVer);
- if (err == null)
- err = fut.validateCache(cctx, recovery, read, null,
keys);
+ if (!affFut.isDone()) {
+ try {
+ affFut.get();
+ }
+ catch (IgniteCheckedException e) {
+ onDone(err);
+
+ return;
+ }
+ }
+
+ for (GridDhtTopologyFuture fut :
cctx.shared().exchange().exchangeFutures()) {
+ if (fut.exchangeDone() &&
fut.topologyVersion().equals(lastChangeVer)) {
+ Throwable err = fut.validateCache(cctx, recovery, read,
null, keys);
if (err != null) {
onDone(err);
@@ -1667,10 +1674,6 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
if (res.compatibleRemapVersion()) {
if (tx != null) {
- // Versions are compatible.
- cctx.shared().exchange().
-
lastAffinityChangedTopologyVersion(res.clientRemapVersion(),
tx.topologyVersionSnapshot());
-
tx.onRemap(res.clientRemapVersion(), false);
// Use remapped version for all subsequent mappings.
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 97535f5..5a4ee46 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
@@ -1647,10 +1647,6 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
if (res.compatibleRemapVersion()) {
if (tx != null) {
- // Versions are compatible.
- cctx.shared().exchange().
-
lastAffinityChangedTopologyVersion(res.clientRemapVersion(),
tx.topologyVersionSnapshot());
-
tx.onRemap(res.clientRemapVersion(), false);
// Use remapped version for all subsequent mappings.
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientSlowDiscoveryAbstractTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientSlowDiscoveryAbstractTest.java
new file mode 100644
index 0000000..24555b5
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientSlowDiscoveryAbstractTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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;
+
+import java.util.function.Supplier;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+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.IgniteInClosure;
+import org.apache.ignite.spi.communication.CommunicationSpi;
+import org.apache.ignite.spi.discovery.DiscoverySpi;
+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.spi.discovery.tcp.messages.TcpDiscoveryNodeAddFinishedMessage;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class ClientSlowDiscoveryAbstractTest extends GridCommonAbstractTest {
+ /** Cache name. */
+ protected static final String CACHE_NAME = "cache";
+
+ /** Cache configuration. */
+ private final CacheConfiguration ccfg = new CacheConfiguration(CACHE_NAME)
+ .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
+ .setReadFromBackup(false)
+ .setBackups(1)
+ .setAffinity(new RendezvousAffinityFunction(false, 64));
+
+ /** Client mode. */
+ protected boolean clientMode;
+
+ /** Communication SPI supplier. */
+ protected Supplier<CommunicationSpi> communicationSpiSupplier =
TestRecordingCommunicationSpi::new;
+
+ /** Discovery SPI supplier. */
+ protected Supplier<DiscoverySpi> discoverySpiSupplier =
TcpDiscoverySpi::new;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setConsistentId(igniteInstanceName);
+ cfg.setCacheConfiguration(ccfg);
+ cfg.setCommunicationSpi(communicationSpiSupplier.get());
+ cfg.setDiscoverySpi(discoverySpiSupplier.get());
+ cfg.setClientMode(clientMode);
+
+ return cfg;
+ }
+
+ /**
+ *
+ */
+ static class NodeJoinInterceptingDiscoverySpi extends TcpDiscoverySpi {
+ /** Interceptor. */
+ protected volatile IgniteInClosure<TcpDiscoveryNodeAddFinishedMessage>
interceptor;
+
+ /** {@inheritDoc} */
+ @Override protected void
startMessageProcess(TcpDiscoveryAbstractMessage msg) {
+ if (msg instanceof TcpDiscoveryNodeAddFinishedMessage &&
interceptor != null)
+ interceptor.apply((TcpDiscoveryNodeAddFinishedMessage) msg);
+ }
+ }
+
+ /**
+ *
+ */
+ static class CustomMessageInterceptingDiscoverySpi extends TcpDiscoverySpi
{
+ /** Interceptor. */
+ protected volatile IgniteInClosure<DiscoveryCustomMessage> interceptor;
+
+ /** {@inheritDoc} */
+ @Override protected void
startMessageProcess(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 (interceptor != null)
+ interceptor.apply(delegate);
+ }
+ }
+}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientDelayedJoinTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientSlowDiscoveryTopologyChangeTest.java
similarity index 56%
rename from
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientDelayedJoinTest.java
rename to
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientSlowDiscoveryTopologyChangeTest.java
index 06aab27..1a13b15 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientDelayedJoinTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientSlowDiscoveryTopologyChangeTest.java
@@ -19,74 +19,25 @@ package org.apache.ignite.internal.processors.cache;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
-import java.util.function.Supplier;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
-import org.apache.ignite.configuration.CacheConfiguration;
-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.discovery.CustomMessageWrapper;
-import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
import
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.spi.communication.CommunicationSpi;
-import org.apache.ignite.spi.discovery.DiscoverySpi;
-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.GridTestUtils;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
/**
- * Tests for delayed client join.
- * When client node joins to cluster it sends SingleMessage to coordinator.
- * During this time topology on server nodes can be changed,
- * because client exchange doesn't require acknowledgement for SingleMessage
on coordinator.
- * Delay is simulated by blocking sending this SingleMessage and resume
sending after topology is changed.
+ *
*/
-public class ClientDelayedJoinTest extends GridCommonAbstractTest {
- /** Cache name. */
- private static final String CACHE_NAME = "cache";
-
- /** Cache configuration. */
- private final CacheConfiguration ccfg = new CacheConfiguration(CACHE_NAME)
- .setReadFromBackup(false)
- .setBackups(1)
- .setAffinity(new RendezvousAffinityFunction(false, 64));
-
- /** Client mode. */
- private boolean clientMode;
-
- /** Communication SPI supplier. */
- private Supplier<CommunicationSpi> communicationSpiSupplier =
TestRecordingCommunicationSpi::new;
-
- /** Discovery SPI supplier. */
- private Supplier<DiscoverySpi> discoverySpiSupplier =
CustomMessageInterceptingDiscoverySpi::new;
-
- /** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
- IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
-
- cfg.setConsistentId(igniteInstanceName);
- cfg.setCacheConfiguration(ccfg);
- cfg.setCommunicationSpi(communicationSpiSupplier.get());
- cfg.setDiscoverySpi(discoverySpiSupplier.get());
- cfg.setClientMode(clientMode);
-
- return cfg;
- }
-
+public class ClientSlowDiscoveryTopologyChangeTest extends
ClientSlowDiscoveryAbstractTest {
/**
*
*/
@@ -108,7 +59,12 @@ public class ClientDelayedJoinTest extends
GridCommonAbstractTest {
}
/**
- *
+ * Test check that client join works well if cache configured on it
stopped on server nodes
+ * but discovery event about cache stop is not delivered to client node
immediately.
+ * When client node joins to cluster it sends SingleMessage to coordinator.
+ * During this time topology on server nodes can be changed,
+ * because client exchange doesn't require acknowledgement for
SingleMessage on coordinator.
+ * Delay is simulated by blocking sending this SingleMessage and resume
sending after topology is changed.
*/
@Test
public void testClientJoinAndCacheStop() throws Exception {
@@ -131,9 +87,9 @@ public class ClientDelayedJoinTest extends
GridCommonAbstractTest {
GridDhtPartitionsSingleMessage singleMsg =
(GridDhtPartitionsSingleMessage) msg;
return Optional.ofNullable(singleMsg.exchangeId())
- .map(GridDhtPartitionExchangeId::topologyVersion)
- .filter(topVer -> topVer.equals(new
AffinityTopologyVersion(4, 0)))
- .isPresent();
+ .map(GridDhtPartitionExchangeId::topologyVersion)
+ .filter(topVer -> topVer.equals(new AffinityTopologyVersion(4,
0)))
+ .isPresent();
});
communicationSpiSupplier = () -> clientCommSpi;
@@ -150,7 +106,7 @@ public class ClientDelayedJoinTest extends
GridCommonAbstractTest {
DynamicCacheChangeBatch cacheChangeBatch =
(DynamicCacheChangeBatch) msg;
boolean hasCacheStopReq = cacheChangeBatch.requests().stream()
- .anyMatch(req -> req.stop() &&
req.cacheName().equals(CACHE_NAME));
+ .anyMatch(req -> req.stop() &&
req.cacheName().equals(CACHE_NAME));
if (hasCacheStopReq)
U.awaitQuiet(clientDiscoSpiBlock);
@@ -185,7 +141,7 @@ public class ClientDelayedJoinTest extends
GridCommonAbstractTest {
}
catch (Exception e) {
assertTrue("Got unexpected exception during cache get " + e,
- X.hasCause(e, CacheStoppedException.class));
+ X.hasCause(e, CacheStoppedException.class));
}
finally {
// Resume processing cache destroy on client node.
@@ -195,7 +151,7 @@ public class ClientDelayedJoinTest extends
GridCommonAbstractTest {
// Wait till cache destroyed on client node.
GridTestUtils.waitForCondition(() -> {
AffinityTopologyVersion topVer =
client.context().cache().context().exchange().lastFinishedFuture()
- .topologyVersion();
+ .topologyVersion();
// Cache destroy version.
return topVer.equals(new AffinityTopologyVersion(4, 1));
@@ -203,37 +159,4 @@ public class ClientDelayedJoinTest extends
GridCommonAbstractTest {
Assert.assertNull("Cache should be destroyed on client node",
client.cache(CACHE_NAME));
}
-
- /**
- *
- */
- static class CustomMessageInterceptingDiscoverySpi extends TcpDiscoverySpi
{
- /** Interceptor. */
- private volatile IgniteInClosure<DiscoveryCustomMessage> interceptor;
-
- /** {@inheritDoc} */
- @Override protected void
startMessageProcess(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 (interceptor != null)
- interceptor.apply(delegate);
- }
- }
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientSlowDiscoveryTransactionRemapTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientSlowDiscoveryTransactionRemapTest.java
new file mode 100644
index 0000000..51a72bc
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/ClientSlowDiscoveryTransactionRemapTest.java
@@ -0,0 +1,473 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import com.google.common.collect.Maps;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.apache.ignite.transactions.TransactionTimeoutException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.mockito.internal.util.collections.Sets;
+
+import static java.util.stream.Collectors.toMap;
+import static java.util.stream.Collectors.toSet;
+import static
org.apache.ignite.testframework.GridTestUtils.assertThrowsWithCause;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static
org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
+
+/**
+ * Tests for client nodes with slow discovery.
+ */
+@RunWith(Parameterized.class)
+public class ClientSlowDiscoveryTransactionRemapTest extends
ClientSlowDiscoveryAbstractTest {
+ /** */
+ @Parameterized.Parameters(name = "isolation = {0}, concurrency = {1},
operation = {2}")
+ public static List<Object[]> parameters() {
+ ArrayList<Object[]> params = new ArrayList<>();
+
+ List<IgniteInClosure<TestTransaction<Integer, Integer>>> operations =
new ArrayList<>();
+
+ operations.add(new NamedClosure<>(putRemoveSameKey,
"putRemoveSameKey"));
+ operations.add(new NamedClosure<>(putRemoveDifferentKey,
"putRemoveDifferentKey"));
+ operations.add(new NamedClosure<>(getPutSameKey, "getPutSameKey"));
+ operations.add(new NamedClosure<>(getPutDifferentKey,
"getPutDifferentKey"));
+ operations.add(new NamedClosure<>(putAllRemoveAllSameKeys,
"putAllRemoveAllSameKeys"));
+ operations.add(new NamedClosure<>(putAllRemoveAllDifferentKeys,
"putAllRemoveAllDifferentKeys"));
+ operations.add(new NamedClosure<>(randomOperation, "random"));
+
+ for (TransactionConcurrency concurrency :
TransactionConcurrency.values()) {
+ for (TransactionIsolation isolation :
TransactionIsolation.values()) {
+ if (!shouldBeTested(concurrency, isolation))
+ continue;
+
+ for (IgniteInClosure<TestTransaction<Integer, Integer>>
operation : operations)
+ params.add(new Object[] {concurrency, isolation,
operation});
+ }
+ }
+
+ return params;
+ }
+
+ /**
+ * @param concurrency Concurrency.
+ * @param isolation Isolation.
+ * @return {@code True} if pair concurrency - isolation should be tested.
+ */
+ private static boolean shouldBeTested(TransactionConcurrency concurrency,
TransactionIsolation isolation) {
+ if (concurrency == PESSIMISTIC)
+ return isolation == REPEATABLE_READ || isolation == READ_COMMITTED;
+ return concurrency == OPTIMISTIC && isolation == SERIALIZABLE;
+ }
+
+ /** Keys set. */
+ private static final int KEYS_SET = 64;
+
+ /** Put remove same key. */
+ private static IgniteInClosure<TestTransaction<Integer, Integer>>
putRemoveSameKey = tx -> {
+ tx.put(1, 1);
+ tx.remove(1);
+ tx.put(1, 100);
+ };
+
+ /** Put remove different key. */
+ private static IgniteInClosure<TestTransaction<Integer, Integer>>
putRemoveDifferentKey = tx -> {
+ tx.put(1, 1);
+ tx.remove(2);
+ };
+
+ /** Get put same key. */
+ private static IgniteInClosure<TestTransaction<Integer, Integer>>
getPutSameKey = tx -> {
+ int val = tx.get(1);
+ tx.put(1, val + 1);
+ };
+
+ /** Get put different key. */
+ private static IgniteInClosure<TestTransaction<Integer, Integer>>
getPutDifferentKey = tx -> {
+ int val = tx.get(1);
+ tx.put(2, val + 1);
+ };
+
+ /** Put all remove all same keys. */
+ private static IgniteInClosure<TestTransaction<Integer, Integer>>
putAllRemoveAllSameKeys = tx -> {
+ tx.putAll(Maps.asMap(Sets.newSet(1, 2, 3, 4, 5), k -> k));
+ tx.removeAll(Sets.newSet(1, 2, 3, 4, 5));
+ };
+
+ /** Put all remove all different keys. */
+ private static IgniteInClosure<TestTransaction<Integer, Integer>>
putAllRemoveAllDifferentKeys = tx -> {
+ tx.putAll(Maps.asMap(Sets.newSet(1, 2, 3, 4, 5), k -> k));
+ tx.removeAll(Sets.newSet(6, 7, 8, 9, 10));
+ };
+
+ /** Random operation. */
+ private static IgniteInClosure<TestTransaction<Integer, Integer>>
randomOperation = tx -> {
+ long seed = ThreadLocalRandom.current().nextLong();
+ log.info("Seed: " + seed);
+ Random random = new Random(seed);
+
+ for (int it = 0; it < 10; it++) {
+ int operation =
random.nextInt(TestTransaction.POSSIBLE_OPERATIONS);
+
+ switch (operation) {
+ // Get:
+ case 0: {
+ int key = random.nextInt(KEYS_SET);
+
+ tx.get(key);
+
+ break;
+ }
+ // Put:
+ case 1: {
+ int key = random.nextInt(KEYS_SET);
+ int val = random.nextInt(KEYS_SET);
+
+ tx.put(key, val);
+
+ break;
+ }
+ // Remove:
+ case 2: {
+ int key = random.nextInt(KEYS_SET);
+
+ tx.remove(key);
+
+ break;
+ }
+ // Put All:
+ case 3: {
+ tx.putAll(
+ random.ints(5, 0, KEYS_SET)
+ .boxed()
+ .distinct()
+ .collect(toMap(
+ k -> k, k -> k)
+ )
+ );
+
+ break;
+ }
+ // Remove All:
+ case 4: {
+ tx.removeAll(
+ random.ints(5, 0, KEYS_SET).boxed().collect(toSet())
+ );
+
+ break;
+ }
+ }
+ }
+ };
+
+ /**
+ * Interface to work with cache operations within transaction.
+ */
+ private static interface TestTransaction<K, V> {
+ /** Possible operations. */
+ static int POSSIBLE_OPERATIONS = 5;
+
+ /**
+ * @param key Key.
+ * @return Value.
+ */
+ V get(K key);
+
+ /**
+ * @param key Key.
+ * @param val Value.
+ */
+ void put(K key, V val);
+
+ /**
+ * @param key Key.
+ */
+ void remove(K key);
+
+ /**
+ * @param map Map.
+ */
+ void putAll(Map<K, V> map);
+
+ /**
+ * @param keys Keys.
+ */
+ void removeAll(Set<K> keys);
+ }
+
+ /**
+ * Closure with possibility to set name to have proper print in test
parameters.
+ */
+ private static class NamedClosure<K, V> implements
IgniteInClosure<TestTransaction<K, V>> {
+ /** Closure. */
+ private final IgniteInClosure<TestTransaction<K, V>> c;
+
+ /** Name. */
+ private final String name;
+
+ /**
+ * @param c Closure.
+ * @param name Name.
+ */
+ public NamedClosure(IgniteInClosure<TestTransaction<K, V>> c, String
name) {
+ this.c = c;
+ this.name = name;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(TestTransaction<K, V> kvTestTransaction) {
+ c.apply(kvTestTransaction);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return name;
+ }
+ }
+
+ /**
+ * Implementation for transaction operations backed by Ignite cache.
+ */
+ private static class TestTransactionEngine<K, V> implements
TestTransaction<K, V> {
+ /** Removed. */
+ private final Object RMV = new Object();
+
+ /** Cache. */
+ private final IgniteCache<K, V> cache;
+
+ /** Map to consistency check. */
+ private final Map<K, Object> map;
+
+ /**
+ * @param cache Cache.
+ */
+ TestTransactionEngine(IgniteCache<K, V> cache) {
+ this.cache = cache;
+ this.map = new HashMap<>();
+ }
+
+ /** {@inheritDoc} */
+ @Override public V get(K key) {
+ return cache.get(key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void put(K key, V val) {
+ map.put(key, val);
+ cache.put(key, val);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void remove(K key) {
+ map.put(key, RMV);
+ cache.remove(key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void putAll(Map<K, V> map) {
+ for (Map.Entry<K, V> entry : map.entrySet())
+ this.map.put(entry.getKey(), entry.getValue());
+ cache.putAll(map);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void removeAll(Set<K> keys) {
+ for (K key : keys)
+ map.put(key, RMV);
+ cache.removeAll(keys);
+ }
+
+ /**
+ * Consistency check for transaction operations.
+ */
+ public void consistencyCheck() {
+ for (Map.Entry<K, Object> entry : map.entrySet()) {
+ if (entry.getValue() == RMV)
+ Assert.assertNull("Value is not null for key: " +
entry.getKey(), cache.get(entry.getKey()));
+ else
+ Assert.assertEquals("Values are different for key: " +
entry.getKey(),
+ entry.getValue(),
+ cache.get(entry.getKey())
+ );
+ }
+ }
+ }
+
+ /** Concurrency. */
+ @Parameterized.Parameter(0)
+ public TransactionConcurrency concurrency;
+
+ /** Isolation. */
+ @Parameterized.Parameter(1)
+ public TransactionIsolation isolation;
+
+ /** Operation. */
+ @Parameterized.Parameter(2)
+ public IgniteInClosure<TestTransaction<?, ?>> operation;
+
+ /** Client disco spi block. */
+ private CountDownLatch clientDiscoSpiBlock;
+
+ /** Client node to perform operations. */
+ private IgniteEx clnt;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ stopAllGrids();
+
+ cleanPersistenceDir();
+
+ startGrid(0);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ cleanPersistenceDir();
+ }
+
+ /** */
+ @Before
+ public void before() throws Exception {
+ clientMode = true;
+
+ NodeJoinInterceptingDiscoverySpi clientDiscoSpi = new
NodeJoinInterceptingDiscoverySpi();
+
+ clientDiscoSpiBlock = new CountDownLatch(1);
+
+ // Delay node join of second client.
+ clientDiscoSpi.interceptor = msg -> {
+ if (msg.nodeId().toString().endsWith("2"))
+ U.awaitQuiet(clientDiscoSpiBlock);
+ };
+
+ discoverySpiSupplier = () -> clientDiscoSpi;
+
+ clnt = startGrid(1);
+
+ for (int k = 0; k < 64; k++)
+ clnt.cache(CACHE_NAME).put(k, 0);
+
+ discoverySpiSupplier = TcpDiscoverySpi::new;
+
+ startGrid(2);
+ }
+
+ /** */
+ @After
+ public void after() throws Exception {
+ // Stop client nodes.
+ stopGrid(1);
+ stopGrid(2);
+ }
+
+ /** */
+ @Test
+ public void testTransactionRemap() throws Exception {
+ TestTransactionEngine engine = new
TestTransactionEngine<>(clnt.cache(CACHE_NAME));
+
+ IgniteInternalFuture<?> txFut = GridTestUtils.runAsync(() -> {
+ try (Transaction tx = clnt.transactions().txStart(concurrency,
isolation)) {
+ operation.apply(engine);
+
+ tx.commit();
+ }
+ });
+
+ try {
+ txFut.get(1, TimeUnit.SECONDS);
+ }
+ catch (IgniteFutureTimeoutCheckedException te) {
+ // Expected.
+ }
+ finally {
+ clientDiscoSpiBlock.countDown();
+ }
+
+ // After resume second client join, transaction should succesfully
await new affinity and commit.
+ txFut.get();
+
+ // Check consistency after transaction commit.
+ engine.consistencyCheck();
+ }
+
+ /** */
+ @Test
+ public void testTransactionRemapWithTimeout() throws Exception {
+ TestTransactionEngine engine = new
TestTransactionEngine<>(clnt.cache(CACHE_NAME));
+
+ IgniteInternalFuture<?> txFut = GridTestUtils.runAsync(() -> {
+ try (Transaction tx = clnt.transactions().txStart(concurrency,
isolation, 1_000, 1_000_000)) {
+ operation.apply(engine);
+
+ tx.commit();
+ }
+ });
+
+ try {
+ txFut.get(2, TimeUnit.SECONDS);
+ }
+ catch (IgniteFutureTimeoutCheckedException te) {
+ // Expected.
+ }
+ finally {
+ clientDiscoSpiBlock.countDown();
+ }
+
+ // After resume second client join, transaction should be timed out
and rolled back.
+ if (concurrency == PESSIMISTIC) {
+ assertThrowsWithCause((Callable<Object>) txFut::get,
TransactionTimeoutException.class);
+
+ // Check that initial data is not changed by rollbacked
transaction.
+ for (int k = 0; k < KEYS_SET; k++)
+ Assert.assertEquals("Cache consistency is broken for key: " +
k, 0, clnt.cache(CACHE_NAME).get(k));
+ }
+ else {
+ txFut.get();
+
+ engine.consistencyCheck();
+ }
+ }
+}
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
index 73024f6..a22e341 100644
---
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
@@ -17,12 +17,10 @@
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;
@@ -31,16 +29,19 @@ 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.IgniteFutureTimeoutCheckedException;
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.GridDhtPartitionExchangeId;
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.preloader.GridDhtPartitionsFullMessage;
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.GridConcurrentSkipListSet;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiPredicate;
@@ -130,7 +131,19 @@ public class TxCrossCacheMapOnInvalidTopologyTest extends
GridCommonAbstractTest
}
/**
+ * Test scenario: cross-cache tx is started when node is left in the
middle of rebalance, first cache is rebalanced
+ * and second is partially rebalanced.
*
+ * First cache map request will trigger client compatible remap for
pessimistic txs,
+ * second cache map request should use new topology version.
+ *
+ * For optimistic tx remap is enforced if more than one mapping in
transaction or all enlisted caches have compatible
+ * assignments.
+ *
+ * Success: tx is finished on ideal topology version over all mapped nodes.
+ *
+ * @param concurrency Concurrency.
+ * @param isolation Isolation.
*/
private void doTestCrossCacheTxMapOnInvalidTopology(TransactionConcurrency
concurrency, TransactionIsolation isolation) throws Exception {
try {
@@ -139,7 +152,7 @@ public class TxCrossCacheMapOnInvalidTopologyTest extends
GridCommonAbstractTest
awaitPartitionMapExchange();
- Ignite client = startGrid("client");
+ IgniteEx client = startGrid("client");
assertNotNull(client.cache(CACHE1));
assertNotNull(client.cache(CACHE2));
@@ -159,9 +172,12 @@ public class TxCrossCacheMapOnInvalidTopologyTest extends
GridCommonAbstractTest
final AffinityTopologyVersion joinVer = new
AffinityTopologyVersion(4, 0);
AffinityTopologyVersion leftVer = new AffinityTopologyVersion(5,
0);
+ AffinityTopologyVersion idealVer = new AffinityTopologyVersion(5,
1);
AtomicReference<Set<Integer>> full = new AtomicReference<>();
+ GridConcurrentSkipListSet<Integer> leftVerParts = new
GridConcurrentSkipListSet<>();
+
crdSpi.blockMessages((node, m) -> {
if (m instanceof GridDhtPartitionSupplyMessage) {
GridDhtPartitionSupplyMessage msg =
(GridDhtPartitionSupplyMessage)m;
@@ -183,7 +199,20 @@ public class TxCrossCacheMapOnInvalidTopologyTest extends
GridCommonAbstractTest
return true;
}
- if (msg.topologyVersion().equals(leftVer))
+ if (msg.topologyVersion().equals(leftVer)) {
+ Map<Integer, Long> last = U.field(msg, "last");
+
+ leftVerParts.addAll(last.keySet());
+
+ return true;
+ }
+ } else if (m instanceof GridDhtPartitionsFullMessage) {
+ GridDhtPartitionsFullMessage msg =
(GridDhtPartitionsFullMessage)m;
+
+ // Delay full message for ideal topology switch.
+ GridDhtPartitionExchangeId exchId = msg.exchangeId();
+
+ if (exchId != null &&
exchId.topologyVersion().equals(idealVer))
return true;
}
@@ -237,76 +266,104 @@ public class TxCrossCacheMapOnInvalidTopologyTest
extends GridCommonAbstractTest
grid(0).cachex(CACHE2).context().affinity().affinityReadyFuture(leftVer).get();
grid(2).cachex(CACHE2).context().affinity().affinityReadyFuture(leftVer).get();
+ AffinityAssignment assignment0 =
grid(0).cachex(CACHE1).context().affinity().assignment(leftVer);
AffinityAssignment assignment =
grid(0).cachex(CACHE2).context().affinity().assignment(leftVer);
// Search for a partition with incompatible assignment.
- int movingPart = -1;
+ int stablePart = -1; // Partition for cache1 which is mapped for
both late and ideal topologies to the same primary.
+ int movingPart = -1; // Partition for cache2 which is mapped for
both late and ideal topologies on different primaries.
- for (int p = 0; p < assignment.assignment().size(); p++) {
- List<ClusterNode> nodes = assignment.assignment().get(p);
- List<ClusterNode> nodes2 = assignment.idealAssignment().get(p);
+ for (int p = 0; p < assignment0.assignment().size(); p++) {
+ List<ClusterNode> curr = assignment.assignment().get(p);
+ List<ClusterNode> ideal = assignment.idealAssignment().get(p);
- if (!nodes.equals(nodes2) && nodes.get(0).order() == 1)
- movingPart = p;
+ if (curr.equals(ideal) && curr.get(0).order() == 1) {
+ stablePart = p;
+
+ break;
+ }
}
- assertFalse(movingPart == -1);
+ assertFalse(stablePart == -1);
- // Delay rebalance for next top ver.
- TestRecordingCommunicationSpi.spi(grid(2)).blockMessages((node,
message) -> {
- if (message instanceof GridDhtPartitionsSingleMessage) {
- GridDhtPartitionsSingleMessage sm =
(GridDhtPartitionsSingleMessage)message;
+ for (int p = 0; p < assignment.assignment().size(); p++) {
+ List<ClusterNode> curr = assignment.assignment().get(p);
+ List<ClusterNode> ideal = assignment.idealAssignment().get(p);
- return sm.exchangeId() != null;
+ if (!curr.equals(ideal) && curr.get(0).order() == 1) {
+ movingPart = p;
+
+ break;
}
+ }
- return false;
- });
+ assertFalse(movingPart == -1);
TestRecordingCommunicationSpi.spi(client).blockMessages(new
IgniteBiPredicate<ClusterNode, Message>() {
- @Override public boolean apply(ClusterNode node, Message
message) {
+ @Override public boolean apply(ClusterNode node, Message msg) {
if (concurrency == PESSIMISTIC)
- return message instanceof GridNearLockRequest;
+ return msg instanceof GridNearLockRequest;
else
- return message instanceof GridNearTxPrepareRequest;
+ return msg instanceof GridNearTxPrepareRequest;
}
});
+ final int finalStablePart = stablePart;
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);
+ IgniteInternalFuture<?> txFut = multithreadedAsync(() -> {
+ try (Transaction tx =
client.transactions().txStart(concurrency, isolation)) {
+ client.cache(CACHE1).put(finalStablePart, 0); // Will map
on crd(order=1).
- client.cache(CACHE1).putAll(map); // Will successfully
lock topology.
- client.cache(CACHE2).put(finalMovingPart, 0); //
Should remap but will go through without fix.
+ // Next request will remap to ideal topology, but it's not
ready on other node except crd.
+ client.cache(CACHE2).put(finalMovingPart, 0);
- tx.commit();
- }
+ tx.commit();
}
}, 1, "tx-thread");
+ // Wait until all missing supply messages are blocked.
+ assertTrue(GridTestUtils.waitForCondition(() ->
leftVerParts.size() == PARTS_CNT - full.get().size(), 5_000));
+
+ // Delay first lock request on late topology.
TestRecordingCommunicationSpi.spi(client).waitForBlocked();
- crdSpi.stopBlock(); // Continue rebalance and trigger next
topology switch.
+ // At this point only supply messages should be blocked.
+ // Unblock to continue rebalance and trigger ideal topology switch.
+ crdSpi.stopBlock(true, null, false, true);
+
+ // Wait until ideal topology is ready on crd.
+
crd.context().cache().context().exchange().affinityReadyFuture(idealVer).get(10_000);
- TestRecordingCommunicationSpi.spi(grid(2)).waitForBlocked();
+ // Other node must wait for full message.
+ assertFalse(GridTestUtils.waitForCondition(() ->
+
grid(2).context().cache().context().exchange().affinityReadyFuture(idealVer).isDone(),
1_000));
+ // Map on unstable topology (PME is in progress on other node).
TestRecordingCommunicationSpi.spi(client).stopBlock();
- doSleep(5000); // Make sure request will listen for current
topology future completion.
+ // Capture local transaction.
+ IgniteInternalTx tx0 =
client.context().cache().context().tm().activeTransactions().iterator().next();
- TestRecordingCommunicationSpi.spi(grid(2)).stopBlock();
+ // Expected behavior: tx must hang (both pessimistic and
optimistic) because topology is not ready.
+ try {
+ txFut.get(3_000);
- crdSpi.stopBlock();
+ fail("TX must not complete");
+ }
+ catch (IgniteFutureTimeoutCheckedException e) {
+ // Expected.
+ }
- awaitPartitionMapExchange();
+ crdSpi.stopBlock();
txFut.get();
+ // Check transaction map version. Should be mapped on ideal
topology.
+ assertEquals(tx0.topologyVersionSnapshot(), idealVer);
+
+ awaitPartitionMapExchange();
+
checkFutures();
}
finally {
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 183f90a..5477a36 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
@@ -469,9 +469,7 @@ public class TxPartitionCounterStateConsistencyTest extends
TxPartitionCounterSt
*/
@Test
public void
testPartitionConsistencyDuringRebalanceAndConcurrentUpdates_NoOp() throws
Exception {
-
testPartitionConsistencyDuringRebalanceConcurrentlyWithTopologyChange(s -> {
- }, s -> {
- });
+
testPartitionConsistencyDuringRebalanceConcurrentlyWithTopologyChange(s -> {},
s -> {});
}
/**
@@ -629,9 +627,9 @@ public class TxPartitionCounterStateConsistencyTest extends
TxPartitionCounterSt
Integer key0 = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME));
Integer key = primaryKey(grid(0).cache(DEFAULT_CACHE_NAME));
- TestRecordingCommunicationSpi spi =
TestRecordingCommunicationSpi.spi(crd);
+ TestRecordingCommunicationSpi crdSpi =
TestRecordingCommunicationSpi.spi(crd);
- spi.blockMessages((node, message) -> {
+ crdSpi.blockMessages((node, message) -> {
if (message instanceof GridDhtPartitionsFullMessage) {
GridDhtPartitionsFullMessage tmp =
(GridDhtPartitionsFullMessage)message;
@@ -644,11 +642,11 @@ public class TxPartitionCounterStateConsistencyTest
extends TxPartitionCounterSt
// Locks mapped wait.
CountDownLatch l = new CountDownLatch(1);
- IgniteInternalFuture fut = GridTestUtils.runAsync(() -> {
+ IgniteInternalFuture startNodeFut = GridTestUtils.runAsync(() -> {
U.awaitQuiet(l);
try {
- startGrid(SERVER_NODES);
+ startGrid(SERVER_NODES); // Start node out of BLT.
}
catch (Exception e) {
fail(X.getFullStackTrace(e));
@@ -662,7 +660,7 @@ public class TxPartitionCounterStateConsistencyTest extends
TxPartitionCounterSt
});
IgniteInternalFuture txFut = GridTestUtils.runAsync(() -> {
- try(Transaction tx = client.transactions().txStart()) {
+ try (Transaction tx = client.transactions().txStart()) {
Map<Integer, Integer> map = new LinkedHashMap<>();
map.put(key, key); // clientFirst=true in lockAll.
@@ -680,7 +678,7 @@ public class TxPartitionCounterStateConsistencyTest extends
TxPartitionCounterSt
l.countDown();
- spi.waitForBlocked(); // Block PME after finish on crd and
wait on others.
+ crdSpi.waitForBlocked(); // Block PME after finish on crd and
wait on others.
cliSpi.stopBlock(); // Start remote lock mapping.
}
@@ -689,12 +687,10 @@ public class TxPartitionCounterStateConsistencyTest
extends TxPartitionCounterSt
}
});
- txFut.get(); // Transaction should not be blocked by concurrent PME.
lockFut.get();
-
- spi.stopBlock();
-
- fut.get();
+ crdSpi.stopBlock();
+ txFut.get();
+ startNodeFut.get();
awaitPartitionMapExchange();
diff --git
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite5.java
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite5.java
index 5a6b244..bb77047 100644
---
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite5.java
+++
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite5.java
@@ -28,6 +28,7 @@ import
org.apache.ignite.cache.affinity.AffinityHistoryCleanupTest;
import org.apache.ignite.cache.affinity.local.LocalAffinityFunctionTest;
import
org.apache.ignite.internal.GridCachePartitionExchangeManagerHistSizeTest;
import
org.apache.ignite.internal.processors.cache.CacheSerializableTransactionsTest;
+import
org.apache.ignite.internal.processors.cache.ClientSlowDiscoveryTransactionRemapTest;
import org.apache.ignite.internal.processors.cache.ClusterReadOnlyModeTest;
import
org.apache.ignite.internal.processors.cache.ClusterStatePartitionedSelfTest;
import
org.apache.ignite.internal.processors.cache.ClusterStateReplicatedSelfTest;
@@ -89,6 +90,7 @@ public class IgniteCacheMvccTestSuite5 {
ignoredTests.add(GridCachePartitionExchangeManagerHistSizeTest.class);
ignoredTests.add(ConcurrentCacheStartTest.class);
ignoredTests.add(IgniteCacheReadThroughEvictionsVariationsSuite.class);
+ ignoredTests.add(ClientSlowDiscoveryTransactionRemapTest.class);
return IgniteCacheTestSuite5.suite(ignoredTests);
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
index 756a913..c1684b0 100644
---
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
+++
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
@@ -32,7 +32,8 @@ import
org.apache.ignite.internal.processors.cache.CacheKeepBinaryTransactionTes
import org.apache.ignite.internal.processors.cache.CacheNearReaderUpdateTest;
import org.apache.ignite.internal.processors.cache.CacheRebalancingSelfTest;
import
org.apache.ignite.internal.processors.cache.CacheSerializableTransactionsTest;
-import org.apache.ignite.internal.processors.cache.ClientDelayedJoinTest;
+import
org.apache.ignite.internal.processors.cache.ClientSlowDiscoveryTopologyChangeTest;
+import
org.apache.ignite.internal.processors.cache.ClientSlowDiscoveryTransactionRemapTest;
import org.apache.ignite.internal.processors.cache.ClusterReadOnlyModeTest;
import
org.apache.ignite.internal.processors.cache.ClusterStatePartitionedSelfTest;
import
org.apache.ignite.internal.processors.cache.ClusterStateReplicatedSelfTest;
@@ -121,7 +122,8 @@ public class IgniteCacheTestSuite5 {
GridTestUtils.addTestIfNeeded(suite,
IgniteCacheReadThroughEvictionsVariationsSuite.class, ignoredTests);
- GridTestUtils.addTestIfNeeded(suite, ClientDelayedJoinTest.class,
ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite,
ClientSlowDiscoveryTopologyChangeTest.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite,
ClientSlowDiscoveryTransactionRemapTest.class, ignoredTests);
//GridTestUtils.addTestIfNeeded(suite,
GridCacheAtomicPreloadSelfTest.class, ignoredTests);
