You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2026/10/21 07:22:44 UTC
(iotdb) 01/01: [Test] Preserve quorum during session seed fallback recovery
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch ty/fix-seed-fallback-it-quorum
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 2ce52b783557a9df515f2f71106f039529e0cc2a
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Wed Oct 21 15:22:16 2026 +0800
[Test] Preserve quorum during session seed fallback recovery
---
.../session/it/SessionSeedNodeFallbackIT.java | 83 ++++++++++++++++------
1 file changed, 63 insertions(+), 20 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/session/it/SessionSeedNodeFallbackIT.java b/integration-test/src/test/java/org/apache/iotdb/session/it/SessionSeedNodeFallbackIT.java
index ceed96c3eb8..8ab83b62335 100644
--- a/integration-test/src/test/java/org/apache/iotdb/session/it/SessionSeedNodeFallbackIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/session/it/SessionSeedNodeFallbackIT.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.it.env.cluster.EnvUtils;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.ClusterIT;
import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.UrlUtils;
import org.apache.iotdb.session.NodesSupplier;
import org.apache.iotdb.session.Session;
@@ -63,6 +64,8 @@ import static org.junit.Assert.assertTrue;
@RunWith(IoTDBTestRunner.class)
@Category(ClusterIT.class)
public class SessionSeedNodeFallbackIT {
+ private static final int RECOVERY_CONNECTION_TIMEOUT_MS = 30_000;
+
@Before
public void setUp() {
EnvFactory.getEnv()
@@ -137,32 +140,41 @@ public class SessionSeedNodeFallbackIT {
return super.scheduleAtFixedRate(command, 1, 1, TimeUnit.DAYS);
}
};
- NodesSupplier business = createSupplier(executor);
- NodesSupplier background = createSupplier(executor);
- try (Session session = new Session.Builder().nodeUrls(urls()).enableAutoFetch(false).build()) {
- Set<TEndPoint> discovered = new HashSet<>(Arrays.asList(endpoint(1), endpoint(2)));
- awaitSnapshot(business, discovered);
- awaitSnapshot(background, discovered);
- session.open(false, 1000, new ConcurrentHashMap<>(), business);
+ // Node 2 keeps a Ratis majority alive for login queries, but is unknown to these clients until
+ // discovery recovers. Initialize a controlled stale snapshot without running discovery first.
+ List<TEndPoint> discovered = Collections.singletonList(endpoint(1));
+ List<TEndPoint> seeds = Arrays.asList(endpoint(0), endpoint(1));
+ NodesSupplier business = createSupplier(executor, discovered, seeds);
+ NodesSupplier background = createSupplier(executor, discovered, seeds);
+ List<String> seedUrls =
+ seeds.stream().map(UrlUtils::convertTEndPointIpv4AndIpv6Url).collect(Collectors.toList());
+ try (Session session =
+ new Session.Builder().nodeUrls(seedUrls).enableAutoFetch(false).build()) {
+ assertEquals(discovered, business.get());
+ assertEquals(discovered, background.get());
+ session.open(false, RECOVERY_CONNECTION_TIMEOUT_MS, new ConcurrentHashMap<>(), business);
+ assertEquals(endpoint(1), connectedEndpoint(session));
EnvFactory.getEnv().startDataNode(0);
awaitStatus(0, NodeStatus.Running);
+ awaitLogin(0);
EnvFactory.getEnv().getDataNodeWrapper(1).stopForcibly();
awaitStatus(1, NodeStatus.Unknown);
- EnvFactory.getEnv().shutdownDataNode(2);
- awaitStatus(2, EnvUtils.getNodeStatusAfterLocalStop());
- assertFalse(business.get().contains(endpoint(0)));
- assertFalse(background.get().contains(endpoint(0)));
+ assertEquals(discovered, business.get());
+ assertEquals(discovered, background.get());
- // Both discovered endpoints are dead. Only the initial seed omitted by discovery is live.
+ // Every discovered endpoint is dead. Node 0 is the only reachable initial seed.
+ Set<TEndPoint> recoveredNodes = new HashSet<>(Arrays.asList(endpoint(0), endpoint(2)));
try (SessionDataSet result = session.executeQueryStatement("SHOW AVAILABLE URLS")) {
- assertEquals(Collections.singleton(endpoint(0)), readEndpoints(result));
+ assertEquals(recoveredNodes, readEndpoints(result));
}
- awaitSnapshot(background, Collections.singleton(endpoint(0)));
- assertEquals(3, background.getSeedNodes().size());
+ assertEquals(endpoint(0), connectedEndpoint(session));
+ awaitSnapshot(background, recoveredNodes);
+ assertEquals(seeds, background.getSeedNodes());
// Wait for the immediate refresh requested by business reconnection, not a 60-second poll.
- executor.submit(() -> {}).get(10, TimeUnit.SECONDS);
- assertEquals(Collections.singletonList(endpoint(0)), business.get());
+ executor.submit(() -> {}).get(60, TimeUnit.SECONDS);
+ assertEquals(recoveredNodes, new HashSet<>(business.get()));
+ assertEquals(seeds, business.getSeedNodes());
} finally {
business.close();
background.close();
@@ -191,9 +203,16 @@ public class SessionSeedNodeFallbackIT {
return (NodesSupplier) field.get(owner);
}
- private NodesSupplier createSupplier(ScheduledThreadPoolExecutor executor) {
- List<TEndPoint> seeds = Arrays.asList(endpoint(0), endpoint(1), endpoint(2));
+ private TEndPoint connectedEndpoint(Session session) throws Exception {
+ Field field = Session.class.getDeclaredField("defaultEndPoint");
+ field.setAccessible(true);
+ return (TEndPoint) field.get(session);
+ }
+
+ private NodesSupplier createSupplier(
+ ScheduledThreadPoolExecutor executor, List<TEndPoint> discovered, List<TEndPoint> seeds) {
return NodesSupplier.createNodeSupplier(
+ discovered,
seeds,
executor,
SessionConfig.DEFAULT_USER,
@@ -201,7 +220,7 @@ public class SessionSeedNodeFallbackIT {
ZoneId.of("UTC"),
SessionConfig.DEFAULT_INITIAL_BUFFER_CAPACITY,
SessionConfig.DEFAULT_MAX_FRAME_SIZE,
- 1000,
+ RECOVERY_CONNECTION_TIMEOUT_MS,
false,
null,
null,
@@ -240,6 +259,30 @@ public class SessionSeedNodeFallbackIT {
Collections.singletonList(status));
}
+ private void awaitLogin(int index) {
+ TEndPoint recovered = endpoint(index);
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(100, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+ // Probe the recovered seed independently so the tested snapshots remain stale.
+ try (Session probe =
+ new Session.Builder()
+ .host(recovered.getIp())
+ .port(recovered.getPort())
+ .enableAutoFetch(false)
+ .build()) {
+ probe.open(false, RECOVERY_CONNECTION_TIMEOUT_MS);
+ try (SessionDataSet result = probe.executeQueryStatement("SHOW AVAILABLE URLS")) {
+ assertTrue(readEndpoints(result).contains(recovered));
+ }
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ throw new AssertionError("Seed is not ready for session queries: " + recovered, e);
+ }
+ });
+ }
+
private void awaitSnapshot(NodesSupplier current, Set<TEndPoint> expected) {
Awaitility.await()
.atMost(30, TimeUnit.SECONDS)