You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sk...@apache.org on 2022/02/22 10:00:19 UTC
[ignite] branch master updated: IGNITE-16589 Fixed an issue that led to failures of server nodes due to short history of affinity assignments. Fixes #9839
This is an automated email from the ASF dual-hosted git repository.
sk0x50 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 6eb90e3 IGNITE-16589 Fixed an issue that led to failures of server nodes due to short history of affinity assignments. Fixes #9839
6eb90e3 is described below
commit 6eb90e3de48137b734faa62001aed2664c48213c
Author: Slava Koptilin <sl...@gmail.com>
AuthorDate: Tue Feb 22 12:59:46 2022 +0300
IGNITE-16589 Fixed an issue that led to failures of server nodes due to short history of affinity assignments. Fixes #9839
---
.../affinity/GridAffinityAssignmentCache.java | 13 +-
.../processors/cache/CacheGroupContext.java | 40 +++--
.../dht/GridDhtAffinityAssignmentResponse.java | 63 ++++++--
.../dht/GridDhtAssignmentFetchFuture.java | 2 +-
.../preloader/GridDhtPartitionsExchangeFuture.java | 24 ++-
.../cache/CacheNoAffinityExchangeTest.java | 163 ++++++++++++++++++++-
6 files changed, 267 insertions(+), 38 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
index c8d5da3..0bc08e2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
@@ -762,6 +762,7 @@ public class GridAffinityAssignmentCache {
/**
* @param topVer Topology version.
* @return Assignment.
+ * @throws IllegalStateException If affinity assignment is not initialized for the given topology version.
*/
public AffinityAssignment readyAffinity(AffinityTopologyVersion topVer) {
AffinityAssignment cache = head.get();
@@ -776,6 +777,7 @@ public class GridAffinityAssignmentCache {
", topVer=" + topVer +
", head=" + head.get().topologyVersion() +
", history=" + affCache.keySet() +
+ ", maxNonShallowHistorySize=" + MAX_NON_SHALLOW_HIST_SIZE +
']');
}
}
@@ -788,6 +790,9 @@ public class GridAffinityAssignmentCache {
*
* @param topVer Topology version.
* @return Cached affinity.
+ * @throws IllegalArgumentException in case of the specified topology version {@code topVer}
+ * is earlier than affinity is calculated
+ * or the history of assignments is already cleaned.
*/
public AffinityAssignment cachedAffinity(AffinityTopologyVersion topVer) {
AffinityTopologyVersion lastAffChangeTopVer =
@@ -802,6 +807,9 @@ public class GridAffinityAssignmentCache {
* @param topVer Topology version for which affinity assignment is requested.
* @param lastAffChangeTopVer Topology version of last affinity assignment change.
* @return Cached affinity.
+ * @throws IllegalArgumentException in case of the specified topology version {@code topVer}
+ * is earlier than affinity is calculated
+ * or the history of assignments is already cleaned.
*/
public AffinityAssignment cachedAffinity(
AffinityTopologyVersion topVer,
@@ -836,17 +844,20 @@ public class GridAffinityAssignmentCache {
", lastAffChangeTopVer=" + lastAffChangeTopVer +
", head=" + head.get().topologyVersion() +
", history=" + affCache.keySet() +
+ ", maxNonShallowHistorySize=" + MAX_NON_SHALLOW_HIST_SIZE +
']');
}
if (cache.topologyVersion().compareTo(topVer) > 0) {
throw new IllegalStateException("Getting affinity for too old topology version that is already " +
- "out of history [locNode=" + ctx.discovery().localNode() +
+ "out of history (try to increase '" + IGNITE_AFFINITY_HISTORY_SIZE + "' system property)" +
+ " [locNode=" + ctx.discovery().localNode() +
", grp=" + cacheOrGrpName +
", topVer=" + topVer +
", lastAffChangeTopVer=" + lastAffChangeTopVer +
", head=" + head.get().topologyVersion() +
", history=" + affCache.keySet() +
+ ", maxNonShallowHistorySize=" + MAX_NON_SHALLOW_HIST_SIZE +
']');
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
index 6cadb65..9cbd8b0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
@@ -1127,22 +1127,38 @@ public class CacheGroupContext {
log.debug("Affinity is ready for topology version, will send response [topVer=" + topVer +
", node=" + nodeId + ']');
- AffinityAssignment assignment = aff.cachedAffinity(topVer);
+ AffinityAssignment assignment;
+ GridDhtAffinityAssignmentResponse res;
+ try {
+ assignment = aff.cachedAffinity(topVer);
- GridDhtAffinityAssignmentResponse res = new GridDhtAffinityAssignmentResponse(
- req.futureId(),
- grpId,
- topVer,
- assignment.assignment());
+ res = new GridDhtAffinityAssignmentResponse(
+ req.futureId(),
+ grpId,
+ topVer,
+ assignment.assignment());
- if (aff.centralizedAffinityFunction()) {
- assert assignment.idealAssignment() != null;
+ if (aff.centralizedAffinityFunction()) {
+ assert assignment.idealAssignment() != null;
- res.idealAffinityAssignment(assignment.idealAssignment());
- }
+ res.idealAffinityAssignment(assignment.idealAssignment());
+ }
- if (req.sendPartitionsState())
- res.partitionMap(top.partitionMap(true));
+ if (req.sendPartitionsState())
+ res.partitionMap(top.partitionMap(true));
+ }
+ catch (IllegalStateException err) {
+ res = new GridDhtAffinityAssignmentResponse(
+ req.futureId(),
+ grpId,
+ topVer,
+ Collections.emptyList());
+
+ res.affinityAssignmentsError(
+ new IgniteCheckedException("Failed to prepare the required affinity assignment " +
+ "[nodeId=" + nodeId + ", topVer=" + topVer + ']',
+ err));
+ }
try {
ctx.io().send(nodeId, res, AFFINITY_POOL);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentResponse.java
index e8b40e9..251cc60 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentResponse.java
@@ -70,6 +70,13 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
/** */
private byte[] partBytes;
+ /** Indicates that getting required affinity assignments has been failed. */
+ @GridDirectTransient
+ private IgniteCheckedException affAssignmentErr;
+
+ /** Serialized error. */
+ private byte[] affAssignmentErrBytes;
+
/**
* Empty constructor.
*/
@@ -215,7 +222,7 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 9;
+ return 10;
}
/**
@@ -233,6 +240,9 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
if (partMap != null && partBytes == null)
partBytes = U.zip(U.marshal(ctx.marshaller(), partMap));
+
+ if (affAssignmentErr != null && affAssignmentErrBytes == null)
+ affAssignmentErrBytes = U.marshal(ctx, affAssignmentErr);
}
/** {@inheritDoc} */
@@ -250,6 +260,27 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
if (partBytes != null && partMap == null)
partMap = U.unmarshalZip(ctx.marshaller(), partBytes, U.resolveClassLoader(ldr, ctx.gridConfig()));
+
+ if (affAssignmentErrBytes != null && affAssignmentErr == null)
+ affAssignmentErr = U.unmarshal(ctx, affAssignmentErrBytes, U.resolveClassLoader(ldr, ctx.gridConfig()));
+ }
+
+ /**
+ * Error that caused failure to get affinity assignments.
+ *
+ * @param err Cause of failure to calculate/get affiniti assignments.
+ */
+ public void affinityAssignmentsError(IgniteCheckedException err) {
+ affAssignmentErr = err;
+ }
+
+ /**
+ * Returns error that caused failure to get affinity assignments.
+ *
+ * @return Error that caused failure to get affinity assignments.
+ */
+ public IgniteCheckedException affinityAssignmentsError() {
+ return affAssignmentErr;
}
/** {@inheritDoc} */
@@ -273,30 +304,36 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
switch (writer.state()) {
case 4:
- if (!writer.writeByteArray("affAssignmentIdsBytes", affAssignmentIdsBytes))
+ if (!writer.writeByteArray("affAssignmentErrBytes", affAssignmentErrBytes))
return false;
writer.incrementState();
case 5:
- if (!writer.writeLong("futId", futId))
+ if (!writer.writeByteArray("affAssignmentIdsBytes", affAssignmentIdsBytes))
return false;
writer.incrementState();
case 6:
- if (!writer.writeByteArray("idealAffAssignmentBytes", idealAffAssignmentBytes))
+ if (!writer.writeLong("futId", futId))
return false;
writer.incrementState();
case 7:
- if (!writer.writeByteArray("partBytes", partBytes))
+ if (!writer.writeByteArray("idealAffAssignmentBytes", idealAffAssignmentBytes))
return false;
writer.incrementState();
case 8:
+ if (!writer.writeByteArray("partBytes", partBytes))
+ return false;
+
+ writer.incrementState();
+
+ case 9:
if (!writer.writeAffinityTopologyVersion("topVer", topVer))
return false;
@@ -319,7 +356,7 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
switch (reader.state()) {
case 4:
- affAssignmentIdsBytes = reader.readByteArray("affAssignmentIdsBytes");
+ affAssignmentErrBytes = reader.readByteArray("affAssignmentErrBytes");
if (!reader.isLastRead())
return false;
@@ -327,7 +364,7 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
reader.incrementState();
case 5:
- futId = reader.readLong("futId");
+ affAssignmentIdsBytes = reader.readByteArray("affAssignmentIdsBytes");
if (!reader.isLastRead())
return false;
@@ -335,7 +372,7 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
reader.incrementState();
case 6:
- idealAffAssignmentBytes = reader.readByteArray("idealAffAssignmentBytes");
+ futId = reader.readLong("futId");
if (!reader.isLastRead())
return false;
@@ -343,7 +380,7 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
reader.incrementState();
case 7:
- partBytes = reader.readByteArray("partBytes");
+ idealAffAssignmentBytes = reader.readByteArray("idealAffAssignmentBytes");
if (!reader.isLastRead())
return false;
@@ -351,6 +388,14 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
reader.incrementState();
case 8:
+ partBytes = reader.readByteArray("partBytes");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 9:
topVer = reader.readAffinityTopologyVersion("topVer");
if (!reader.isLastRead())
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAssignmentFetchFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAssignmentFetchFuture.java
index cb15227..e2fea81 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAssignmentFetchFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAssignmentFetchFuture.java
@@ -153,7 +153,7 @@ public class GridDhtAssignmentFetchFuture extends GridFutureAdapter<GridDhtAffin
}
if (res0 != null)
- onDone(res);
+ onDone(res, res0.affinityAssignmentsError());
}
/**
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 4ef0d91..3e49fbf 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
@@ -4410,15 +4410,25 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
Collection<Integer> affReq = msg.cacheGroupsAffinityRequest();
if (affReq != null) {
- Map<Integer, CacheGroupAffinityMessage> cachesAff = U.newHashMap(affReq.size());
+ try {
+ Map<Integer, CacheGroupAffinityMessage> cachesAff = U.newHashMap(affReq.size());
- CacheGroupAffinityMessage.createAffinityMessages(
- cctx,
- finishState.resTopVer,
- affReq,
- cachesAff);
+ CacheGroupAffinityMessage.createAffinityMessages(
+ cctx,
+ finishState.resTopVer,
+ affReq,
+ cachesAff);
- fullMsg.joinedNodeAffinity(cachesAff);
+ fullMsg.joinedNodeAffinity(cachesAff);
+ }
+ catch (IllegalStateException e) {
+ // Cannot create affinity message.
+ Map<UUID, Exception> errs = Collections.singletonMap(
+ nodeId,
+ node.isClient() ? new IgniteNeedReconnectException(node, e) : new IgniteCheckedException(e));
+
+ fullMsg.setErrorsMap(errs);
+ }
}
if (!fullMsg.exchangeId().equals(msg.exchangeId())) {
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheNoAffinityExchangeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheNoAffinityExchangeTest.java
index 4dfc918..5292de5 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheNoAffinityExchangeTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheNoAffinityExchangeTest.java
@@ -18,12 +18,15 @@
package org.apache.ignite.internal.processors.cache;
import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteInterruptedException;
import org.apache.ignite.IgniteSystemProperties;
@@ -36,12 +39,17 @@ 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.failure.AbstractFailureHandler;
+import org.apache.ignite.failure.FailureContext;
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.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAffinityAssignmentRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessageV2;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
import org.apache.ignite.internal.util.typedef.G;
+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.spi.discovery.tcp.TcpDiscoverySpi;
@@ -58,6 +66,10 @@ import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.ignite.transactions.Transaction;
import org.junit.Test;
+import static org.apache.ignite.cluster.ClusterState.ACTIVE;
+import static org.apache.ignite.cluster.ClusterState.INACTIVE;
+import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
+
/**
*
*/
@@ -75,6 +87,12 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
private final TcpDiscoveryIpFinder CLIENT_IP_FINDER = new TcpDiscoveryVmIpFinder()
.setAddresses(Collections.singleton("127.0.0.1:47500"));
+ /** Custom communication SPI that can be used by the client node. */
+ private volatile TestRecordingCommunicationSpi clientCommSpi;
+
+ /** Stores errors that triggered failure handler. */
+ private final Map<String, Throwable> errs = new ConcurrentHashMap<>();
+
/** {@inheritDoc} */
@Override protected void beforeTestsStarted() throws Exception {
MvccFeatureChecker.skipIfNotSupported(MvccFeatureChecker.Feature.ENTRY_LOCK);
@@ -90,12 +108,16 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
cfg.setDiscoverySpi(new TestDiscoverySpi().setIpFinder(IP_FINDER));
- cfg.setActiveOnStart(false);
+ cfg.setClusterStateOnStart(INACTIVE);
cfg.setDataStorageConfiguration(new DataStorageConfiguration().setDefaultDataRegionConfiguration(
new DataRegionConfiguration().setMaxSize(200 * 1024 * 1024)));
if (cfg.isClientMode()) {
+ TestRecordingCommunicationSpi customSpi = clientCommSpi;
+ if (customSpi != null)
+ cfg.setCommunicationSpi(customSpi);
+
// It is necessary to ensure that client always connects to grid(0).
((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(CLIENT_IP_FINDER);
@@ -111,6 +133,14 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
}
}
+ cfg.setFailureHandler(new AbstractFailureHandler() {
+ /** {@inheritDoc} */
+ @Override protected boolean handle(Ignite ignite, FailureContext failureCtx) {
+ errs.put(ignite.configuration().getIgniteInstanceName(), failureCtx.error());
+ return false;
+ }
+ });
+
return cfg;
}
@@ -130,7 +160,7 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
public void testNoAffinityChangeOnClientJoin() throws Exception {
Ignite ig = startGrids(4);
- ig.cluster().active(true);
+ ig.cluster().state(ACTIVE);
IgniteCache<Integer, Integer> atomicCache = ig.createCache(new CacheConfiguration<Integer, Integer>()
.setName("atomic").setAtomicityMode(CacheAtomicityMode.ATOMIC));
@@ -186,7 +216,7 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
public void testNoAffinityChangeOnClientLeft() throws Exception {
Ignite ig = startGrids(4);
- ig.cluster().active(true);
+ ig.cluster().state(ACTIVE);
IgniteCache<Integer, Integer> atomicCache = ig.createCache(new CacheConfiguration<Integer, Integer>()
.setName("atomic").setAtomicityMode(CacheAtomicityMode.ATOMIC));
@@ -247,7 +277,7 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
try {
Ignite ig = startGridsMultiThreaded(4);
- ig.cluster().active(true);
+ ig.cluster().state(ACTIVE);
IgniteCache<Integer, Integer> atomicCache = ig.createCache(new CacheConfiguration<Integer, Integer>()
.setName("atomic").setAtomicityMode(CacheAtomicityMode.ATOMIC).setCacheMode(CacheMode.REPLICATED));
@@ -341,7 +371,7 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
public void doTestMulipleClientLeaveJoin() throws Exception {
Ignite ig = startGrids(2);
- ig.cluster().active(true);
+ ig.cluster().state(ACTIVE);
IgniteEx stableClient = startClientGrid(2);
@@ -354,7 +384,7 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
awaitPartitionMapExchange();
- IgniteInternalFuture fut = GridTestUtils.runAsync(new Runnable() {
+ IgniteInternalFuture<?> fut = GridTestUtils.runAsync(new Runnable() {
@Override public void run() {
for (int i = 0; i < 10; i++) {
try {
@@ -371,7 +401,7 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
CountDownLatch clientTxLatch = new CountDownLatch(1);
- IgniteInternalFuture loadFut = GridTestUtils.runAsync(new Runnable() {
+ IgniteInternalFuture<?> loadFut = GridTestUtils.runAsync(new Runnable() {
@Override public void run() {
try (Transaction tx = stableClient.transactions().txStart()) {
ThreadLocalRandom r = ThreadLocalRandom.current();
@@ -404,7 +434,7 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
public void testAffinityChangeOnClientConnectWithStaticallyConfiguredCaches() throws Exception {
Ignite ig = startGrids(2);
- ig.cluster().active(true);
+ ig.cluster().state(ACTIVE);
TestDiscoverySpi discoSpi = (TestDiscoverySpi)grid(1).context().discovery().getInjectedDiscoverySpi();
@@ -448,6 +478,123 @@ public class CacheNoAffinityExchangeTest extends GridCommonAbstractTest {
}
/**
+ * Tests the "long" joining of the client node to the cluster
+ * when the message from the client is handled after affinity assignments are wiped from the history.
+ *
+ * Expected result: client node should reconnect to the cluster, server nodes should not be affected in any way.
+ *
+ * @throws Exception If Failed.
+ */
+ @Test
+ @WithSystemProperty(key = IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE, value = "2")
+ public void testNoAffinityOnJoiningClientNode() throws Exception {
+ Ignite ig = startGrids(1);
+
+ ig.cluster().state(ACTIVE);
+
+ clientCommSpi = new TestRecordingCommunicationSpi();
+
+ clientCommSpi.blockMessages(
+ GridDhtPartitionsSingleMessage.class,
+ ig.configuration().getIgniteInstanceName());
+
+ // Start client node and block initial PME in order to guarantee that single message will be processed
+ // when affinity history for the corresponding topology version is already cleaned.
+ IgniteInternalFuture<?> startClientFut = GridTestUtils.runAsync(() -> {
+ try {
+ startClientCaches = true;
+
+ startClientGrid(1);
+ }
+ catch (Exception e) {
+ throw new RuntimeException("Unexpected exception.", e);
+ }
+ });
+
+ // Wait for initial PME.
+ clientCommSpi.waitForBlocked();
+
+ startClientCaches = false;
+
+ // Start new server nodes in order to clean the history of affinity assignments.
+ for (int i = 2; i < 5; i++) {
+ startGrid(i);
+ stopGrid(i, true);
+ }
+
+ // Send single message from the client.
+ clientCommSpi.stopBlock();
+
+ // Make sure that client node successfully started.
+ startClientFut.get(getTestTimeout());
+
+ awaitPartitionMapExchange();
+
+ StringBuilder failures = new StringBuilder();
+ errs.forEach((k, v) -> failures.append("nodeId=").append(k).append(", err=").append(v).append(U.nl()));
+
+ assertTrue(
+ "Failure handler should not be triggered " + failures.toString(),
+ errs.isEmpty());
+ }
+
+ /**
+ * Tests getting a cache on the client node when the history of affinity assignments is not enough.
+ *
+ * Expected result: client request should be rejected with the cause, server nodes should not be affected in any way.
+ *
+ * @throws Exception If Failed.
+ */
+ @Test
+ @WithSystemProperty(key = IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE, value = "2")
+ public void testNoAffinityOnClientCacheStart() throws Exception {
+ Ignite ig = startGrids(1);
+
+ ig.cluster().state(ACTIVE);
+
+ ig.getOrCreateCache("client-cache");
+
+ startClientCaches = false;
+
+ IgniteEx client = startClientGrid(1);
+ TestRecordingCommunicationSpi clientSpi = TestRecordingCommunicationSpi.spi(client);
+ clientSpi.blockMessages((node, msg) -> msg instanceof GridDhtAffinityAssignmentRequest);
+
+ // Block creating client cache in order to guarantee that GridDhtAffinityAssignmentRequest message
+ // will be processed when affinity history for the corresponding topology version is already cleaned.
+ IgniteInternalFuture<?> startClientCacheFut = GridTestUtils.runAsync(() -> {
+ client.cache("client-cache");
+ });
+
+ clientSpi.waitForBlocked();
+
+ startClientCaches = false;
+
+ // Start new server nodes in order to clean the history of affinity assignments.
+ for (int i = 2; i < 5; i++) {
+ startGrid(i);
+ stopGrid(i, true);
+ }
+
+ // Send GridDhtAffinityAssignmentRequest from the client.
+ clientSpi.stopBlock();
+
+ assertThrows(log, () -> startClientCacheFut.get(), IgniteCheckedException.class, null);
+
+ awaitPartitionMapExchange();
+
+ StringBuilder failures = new StringBuilder();
+ errs.forEach((k, v) -> failures.append("nodeId=").append(k).append(", err=").append(v).append(U.nl()));
+
+ assertTrue(
+ "Failure handler should not be triggered " + failures.toString(),
+ errs.isEmpty());
+
+ // Make sure that the next call is successful.
+ assertNotNull(client.cache("client-cache"));
+ }
+
+ /**
*
*/
public static class TestDiscoverySpi extends TcpDiscoverySpi {