This is an automated email from the ASF dual-hosted git repository.
sergey-chugunov-1985 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 5c899284ac5 IGNITE-28855 Client node permanently loses discovery
events delivered right after the reconnect response (#13327)
5c899284ac5 is described below
commit 5c899284ac5cc9e3739cfd96bc8a8686a319ff04
Author: Anton Vinogradov <[email protected]>
AuthorDate: Fri Jul 31 16:08:54 2026 +0300
IGNITE-28855 Client node permanently loses discovery events delivered right
after the reconnect response (#13327)
---
.../ignite/spi/discovery/tcp/ClientImpl.java | 25 +++-
.../tcp/TcpDiscoveryClientTopologyGapTest.java | 135 +++++++++++++++++++++
.../IgniteSpiDiscoverySelfTestSuite2.java | 2 +
3 files changed, 157 insertions(+), 5 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
index 918a29bab42..43d4f1b4867 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
@@ -776,7 +776,7 @@ class ClientImpl extends TcpDiscoveryImpl {
log.debug("Message has been sent to address [msg=" + msg +
", addr=" + addr +
", rmtNodeId=" + rmtNodeId + ']');
- return new T2<>(new SocketStream(sock), spi.readReceipt(sock,
timeoutHelper.nextTimeoutChunk(ackTimeout0)));
+ return new T2<>(new SocketStream(sock, ses),
spi.readReceipt(sock, timeoutHelper.nextTimeoutChunk(ackTimeout0)));
}
catch (IOException | IgniteCheckedException e) {
U.closeQuiet(sock);
@@ -1162,7 +1162,7 @@ class ClientImpl extends TcpDiscoveryImpl {
+ ":" + sockStream.sock.getPort());
try {
- TcpDiscoveryIoSession ses = createSession(sock);
+ TcpDiscoveryIoSession ses = sockStream.session();
assert sock.getKeepAlive() && sock.getTcpNoDelay() :
"Socket wasn't configured properly:" +
" KeepAlive " + sock.getKeepAlive() +
@@ -1573,7 +1573,7 @@ class ClientImpl extends TcpDiscoveryImpl {
sockStream = joinRes;
Socket sock = sockStream.socket();
- TcpDiscoveryIoSession ses = createSession(sock);
+ TcpDiscoveryIoSession ses = sockStream.session();
if (isInterrupted())
throw new InterruptedException();
@@ -2709,14 +2709,23 @@ class ClientImpl extends TcpDiscoveryImpl {
/** */
private final Socket sock;
+ /**
+ * The only session ever used to read messages from the socket. Shared
by all socket users
+ * ({@link Reconnector}, {@link SocketReader}), otherwise messages
buffered by one session's
+ * read-ahead would be lost when another session takes the socket over.
+ */
+ private final TcpDiscoveryIoSession ses;
+
/**
* @param sock Socket.
- * @throws IOException If failed to create stream.
+ * @param ses Session bound to the socket.
*/
- public SocketStream(Socket sock) throws IOException {
+ SocketStream(Socket sock, TcpDiscoveryIoSession ses) {
assert sock != null;
+ assert ses != null;
this.sock = sock;
+ this.ses = ses;
}
/**
@@ -2724,7 +2733,13 @@ class ClientImpl extends TcpDiscoveryImpl {
*/
Socket socket() {
return sock;
+ }
+ /**
+ * @return Session bound to the socket.
+ */
+ TcpDiscoveryIoSession session() {
+ return ses;
}
/** {@inheritDoc} */
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryClientTopologyGapTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryClientTopologyGapTest.java
new file mode 100644
index 00000000000..ba86e441f2f
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryClientTopologyGapTest.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.spi.discovery.tcp;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.Ignite;
+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.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+/**
+ * Reproducer for a client node losing a discovery event (topology version
gap) while its router and
+ * the coordinator are sequentially stopped. The gap manifests as {@code
AssertionError: lastVer=N, newVer=N+2}
+ * in {@code ClientImpl.updateTopologyHistory}.
+ */
+public class TcpDiscoveryClientTopologyGapTest extends GridCommonAbstractTest {
+ /** */
+ private static final int SRVS = 5;
+
+ /** */
+ private static final int CLIENTS = 4;
+
+ /** */
+ private static final int ITERS = 300;
+
+ /** Critical failures captured on nodes, by instance name. */
+ private static final Map<String, Throwable> failures = new
ConcurrentHashMap<>();
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setLocalPort(48700);
+
+ cfg.setFailureHandler(new AbstractFailureHandler() {
+ @Override protected boolean handle(Ignite ignite, FailureContext
failureCtx) {
+ failures.put(igniteInstanceName, failureCtx.error());
+
+ return false;
+ }
+ });
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+
+ /**
+ * Mirrors the topology dance of {@code
IgniteCachePartitionLossPolicySelfTest} with 5 nodes and
+ * {@code stopNodes={1, 0, 2}}: grid(0) is both the clients' router and
the coordinator, grid(2)
+ * is the next coordinator. Clients must observe every topology version.
+ */
+ @Test
+ public void testClientSeesAllTopologyVersionsOnSequentialServerStop()
throws Exception {
+ for (int iter = 0; iter < ITERS; iter++) {
+ log.info(">>> Iteration: " + iter);
+
+ failures.clear();
+
+ startGrids(SRVS);
+
+ IgniteEx[] clients = new IgniteEx[CLIENTS];
+
+ for (int i = 0; i < CLIENTS; i++)
+ clients[i] = startClientGrid(SRVS + i);
+
+ stopGrid(1, true);
+ stopGrid(0, true);
+ stopGrid(2, true);
+
+ long expTopVer = SRVS + CLIENTS + 3;
+
+ // Failure detection may push the topology past the expected
version, so it is only a lower bound here,
+ // while the invariant the test is about is checked below.
+ for (IgniteEx client : clients) {
+ if (!GridTestUtils.waitForCondition(() ->
client.cluster().topologyVersion() >= expTopVer, 15_000)) {
+ fail("Client stuck [name=" + client.name() +
+ ", topVer=" + client.cluster().topologyVersion() +
+ ", expTopVer=" + expTopVer +
+ ", failures=" + failures + ']');
+ }
+ }
+
+ for (IgniteEx client : clients)
+ assertNoTopologyGap(client, expTopVer);
+
+ assertTrue("Critical failures detected: " + failures,
failures.isEmpty());
+
+ stopAllGrids();
+ }
+ }
+
+ /**
+ * Checks that the node was notified of every topology version up to
{@code lastVer}. Versions preceding the
+ * node's own join are missing from its history and are skipped.
+ *
+ * @param node Node to check.
+ * @param lastVer Last topology version to check.
+ */
+ private void assertNoTopologyGap(IgniteEx node, long lastVer) {
+ boolean joined = false;
+
+ for (long ver = 1; ver <= lastVer; ver++) {
+ if (node.context().discovery().topology(ver) != null)
+ joined = true;
+ else if (joined)
+ fail("Client lost topology version [name=" + node.name() + ",
ver=" + ver + ']');
+ }
+ }
+}
diff --git
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite2.java
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite2.java
index c35d838d055..7d827a4d3ac 100644
---
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite2.java
+++
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite2.java
@@ -33,6 +33,7 @@ import
org.apache.ignite.spi.discovery.tcp.MultiDataCenterSplitTest;
import
org.apache.ignite.spi.discovery.tcp.TcpClientDiscoverySpiCoordinatorChangeTest;
import org.apache.ignite.spi.discovery.tcp.TcpClientDiscoverySpiSelfTest;
import
org.apache.ignite.spi.discovery.tcp.TcpClientDiscoveryUnresolvedHostTest;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryClientTopologyGapTest;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryConcurrentStartTest;
import
org.apache.ignite.spi.discovery.tcp.TcpDiscoveryDeadNodeAddressResolvingTest;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryIpFinderFailureTest;
@@ -74,6 +75,7 @@ import org.junit.runners.Suite;
LongClientConnectToClusterTest.class,
TcpClientDiscoverySpiCoordinatorChangeTest.class,
TcpClientDiscoveryUnresolvedHostTest.class,
+ TcpDiscoveryClientTopologyGapTest.class,
TcpDiscoveryNodeConsistentIdSelfTest.class,
TcpDiscoveryNodeConfigConsistentIdSelfTest.class,
TcpDiscoveryRestartTest.class,
