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 4eeb4591ab6 IGNITE-21266 Java Thin Client: Fixed PA after cluster
restart with the same topology version. (#11191)
4eeb4591ab6 is described below
commit 4eeb4591ab67bf7c0f9d7c8f738e9a0557ae1561
Author: Mikhail Petrov <32207922+petrov...@users.noreply.github.com>
AuthorDate: Thu Feb 22 12:03:20 2024 +0300
IGNITE-21266 Java Thin Client: Fixed PA after cluster restart with the same
topology version. (#11191)
---
.../client/thin/ClientCacheAffinityContext.java | 33 ++++++++++++++++++++--
.../internal/client/thin/ReliableChannel.java | 19 ++++++++++++-
...ientPartitionAwarenessUnstableTopologyTest.java | 27 +++++++++++++-----
3 files changed, 68 insertions(+), 11 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientCacheAffinityContext.java
b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientCacheAffinityContext.java
index bcd681796da..de42b6532d3 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientCacheAffinityContext.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientCacheAffinityContext.java
@@ -28,6 +28,7 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
+import java.util.function.Predicate;
import org.apache.ignite.IgniteBinary;
import org.apache.ignite.client.ClientPartitionAwarenessMapper;
import org.apache.ignite.client.ClientPartitionAwarenessMapperFactory;
@@ -36,6 +37,8 @@ import org.apache.ignite.internal.util.GridConcurrentHashSet;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.jetbrains.annotations.Nullable;
+import static
org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion.NONE;
+
/**
* Client cache partition awareness context.
*/
@@ -68,13 +71,21 @@ public class ClientCacheAffinityContext {
/** Caches that have been requested partition mappings for. */
private volatile CacheMappingRequest rq;
+ /** Predicate to check whether a connection to the node with the specified
ID is open. */
+ private final Predicate<UUID> connectionEstablishedPredicate;
+
/**
* @param binary Binary data processor.
* @param factory Factory for caches with custom affinity.
*/
- public ClientCacheAffinityContext(IgniteBinary binary, @Nullable
ClientPartitionAwarenessMapperFactory factory) {
+ public ClientCacheAffinityContext(
+ IgniteBinary binary,
+ @Nullable ClientPartitionAwarenessMapperFactory factory,
+ Predicate<UUID> connectionEstablishedPredicate
+ ) {
this.paMapFactory = factory;
this.binary = binary;
+ this.connectionEstablishedPredicate = connectionEstablishedPredicate;
}
/**
@@ -88,7 +99,7 @@ public class ClientCacheAffinityContext {
while (true) {
TopologyNodes lastTop = this.lastTop.get();
- if (lastTop == null || topVer.compareTo(lastTop.topVer) > 0) {
+ if (isTopologyOutdated(lastTop, topVer)) {
if (this.lastTop.compareAndSet(lastTop, new
TopologyNodes(topVer, nodeId)))
return true;
}
@@ -262,7 +273,7 @@ public class ClientCacheAffinityContext {
protected ClientCacheAffinityMapping currentMapping() {
TopologyNodes top = lastTop.get();
- if (top == null)
+ if (isTopologyOutdated(top, NONE))
return null;
ClientCacheAffinityMapping mapping = affinityMapping;
@@ -303,6 +314,22 @@ public class ClientCacheAffinityContext {
}
}
+ /** */
+ private boolean isTopologyOutdated(TopologyNodes top,
AffinityTopologyVersion srvSideTopVer) {
+ if (top == null)
+ return true;
+
+ if (srvSideTopVer.compareTo(top.topVer) > 0)
+ return true;
+
+ for (UUID topNode : top.nodes) {
+ if (connectionEstablishedPredicate.test(topNode))
+ return false;
+ }
+
+ return true;
+ }
+
/**
* Holder for list of nodes for topology version.
*/
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
index f88a90e654e..f3ae29c0e1d 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
@@ -139,7 +139,12 @@ final class ReliableChannel implements AutoCloseable {
partitionAwarenessEnabled = clientCfg.isPartitionAwarenessEnabled();
- affinityCtx = new ClientCacheAffinityContext(binary,
clientCfg.getPartitionAwarenessMapperFactory());
+ affinityCtx = new ClientCacheAffinityContext(
+ binary,
+ clientCfg.getPartitionAwarenessMapperFactory(),
+ this::isConnectionEstablished
+ );
+
discoveryCtx = new ClientDiscoveryContext(clientCfg);
connMgr = new GridNioClientConnectionMultiplexer(clientCfg);
@@ -958,6 +963,18 @@ final class ReliableChannel implements AutoCloseable {
return affinityCtx;
}
+ /** */
+ private boolean isConnectionEstablished(UUID node) {
+ ClientChannelHolder chHolder = nodeChannels.get(node);
+
+ if (chHolder == null || chHolder.isClosed())
+ return false;
+
+ ClientChannel ch = chHolder.ch;
+
+ return ch != null && !ch.closed();
+ }
+
/**
* Channels holder.
*/
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ThinClientPartitionAwarenessUnstableTopologyTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ThinClientPartitionAwarenessUnstableTopologyTest.java
index d2ba0e7e9a1..fe49303bbeb 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ThinClientPartitionAwarenessUnstableTopologyTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ThinClientPartitionAwarenessUnstableTopologyTest.java
@@ -42,6 +42,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import static java.util.stream.IntStream.range;
import static org.apache.ignite.testframework.GridTestUtils.getFieldValue;
import static org.apache.ignite.testframework.GridTestUtils.setFieldValue;
import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
@@ -219,12 +220,25 @@ public class
ThinClientPartitionAwarenessUnstableTopologyTest extends ThinClient
* Test that partition awareness works when reconnecting to the new
cluster (with lower topology version)
*/
@Test
- public void testPartitionAwarenessOnClusterRestart() throws Exception {
- startGrids(3);
+ public void
testPartitionAwarenessOnClusterRestartWithLowerTopologyVersion() throws
Exception {
+ doPartitionAwarenessOnClusterRestartTest(3, 2);
+ }
+
+ /**
+ * Test that partition awareness works when reconnecting to the new
cluster (with the same topology version)
+ */
+ @Test
+ public void
testPartitionAwarenessOnClusterRestartWithSameTopologyVersion() throws
Exception {
+ doPartitionAwarenessOnClusterRestartTest(3, 3);
+ }
+
+ /** */
+ private void doPartitionAwarenessOnClusterRestartTest(int
initialClusterSize, int restartedClusterSize) throws Exception {
+ startGrids(initialClusterSize);
awaitPartitionMapExchange();
- initClient(getClientConfiguration(0, 1, 2), 0, 1, 2);
+ initClient(getClientConfiguration(range(0,
initialClusterSize).toArray()), range(0, initialClusterSize).toArray());
// Test partition awareness before cluster restart.
testPartitionAwareness(true);
@@ -233,17 +247,16 @@ public class
ThinClientPartitionAwarenessUnstableTopologyTest extends ThinClient
Arrays.fill(channels, null);
- // Start 2 grids, so topology version of the new cluster will be less
then old cluster.
- startGrids(2);
+ startGrids(restartedClusterSize);
awaitPartitionMapExchange();
// Send any request to failover.
- client.cache(REPL_CACHE_NAME).put(0, 0);
+ client.cache(cacheName).put(0, 0);
detectTopologyChange();
- awaitChannelsInit(0, 1);
+ awaitChannelsInit(range(0, restartedClusterSize).toArray());
testPartitionAwareness(true);
}
