From c60059332147949fa1347fe1bbcdad3812ecf096 Mon Sep 17 00:00:00 2001 From: JackieTien97 Date: Tue, 20 Oct 2026 09:43:50 +0800 Subject: [PATCH 1/3] Filter offline DataNode URLs and fall back to seed nodes --- .../session/it/SessionSeedNodeFallbackIT.java | 244 +++++++++++ .../apache/iotdb/isession/INodeSupplier.java | 11 + .../apache/iotdb/session/NodesSupplier.java | 83 +++- .../org/apache/iotdb/session/Session.java | 18 +- .../iotdb/session/SessionConnection.java | 29 +- .../iotdb/session/pool/SessionPool.java | 3 +- .../session/SessionSeedNodeFallbackTest.java | 411 ++++++++++++++++++ .../metadata/ShowAvailableUrlsTask.java | 6 +- .../metadata/ShowAvailableUrlsTaskTest.java | 78 ++++ 9 files changed, 840 insertions(+), 43 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/session/it/SessionSeedNodeFallbackIT.java create mode 100644 iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTaskTest.java 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 new file mode 100644 index 0000000000000..81df4e4fcf114 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/session/it/SessionSeedNodeFallbackIT.java @@ -0,0 +1,244 @@ +/* + * 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.iotdb.session.it; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.isession.SessionConfig; +import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.isession.pool.SessionDataSetWrapper; +import org.apache.iotdb.it.env.EnvFactory; +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.UrlUtils; +import org.apache.iotdb.session.NodesSupplier; +import org.apache.iotdb.session.Session; +import org.apache.iotdb.session.pool.SessionPool; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.lang.reflect.Field; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +@RunWith(IoTDBTestRunner.class) +@Category(ClusterIT.class) +public class SessionSeedNodeFallbackIT { + @Before + public void setUp() { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setSchemaReplicationFactor(3) + .setDataReplicationFactor(3) + .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) + .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + EnvFactory.getEnv().initClusterEnvironment(1, 3); + } + + @After + public void tearDown() { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testQueriesDuringRollingRestart() throws Exception { + SessionPool pool = + new SessionPool.Builder().nodeUrls(urls()).maxSize(1).connectionTimeoutInMs(1000).build(); + Session session = new Session.Builder().nodeUrls(urls()).build(); + try { + // Initial Ratis region creation can exceed one second; do not retry its DDL prematurely. + session.open(false, 30_000); + session.executeNonQueryStatement("CREATE DATABASE root.seed_fallback"); + session.executeNonQueryStatement( + "CREATE TIMESERIES root.seed_fallback.d.s WITH DATATYPE=INT32, ENCODING=RLE"); + session.executeNonQueryStatement("INSERT INTO root.seed_fallback.d(time,s) VALUES(1,42)"); + assertQuery(session, pool); + NodesSupplier sessionSupplier = supplier(session); + NodesSupplier poolSupplier = supplier(pool); + for (int index = 0; index < 3; index++) { + EnvFactory.getEnv().shutdownDataNode(index); + awaitStatus(index, EnvUtils.getNodeStatusAfterLocalStop()); + sessionSupplier.run(); + poolSupplier.run(); + assertFalse(sessionSupplier.get().contains(endpoint(index))); + assertFalse(poolSupplier.get().contains(endpoint(index))); + assertQuery(session, pool); + + EnvFactory.getEnv().startDataNode(index); + awaitStatus(index, NodeStatus.Running); + sessionSupplier.run(); + poolSupplier.run(); + assertTrue(sessionSupplier.get().contains(endpoint(index))); + assertTrue(poolSupplier.get().contains(endpoint(index))); + assertQuery(session, pool); + } + } finally { + try { + session.close(); + } catch (IoTDBConnectionException e) { + // Cached connections may still belong to a stopped process. Only cleanup tolerates this; + // all query and discovery assertions above must succeed without ignoring failures. + } finally { + pool.close(); + } + } + } + + @Test + public void testBusinessAndBackgroundRecoveryThroughOmittedSeed() throws Exception { + EnvFactory.getEnv().shutdownDataNode(0); + awaitStatus(0, EnvUtils.getNodeStatusAfterLocalStop()); + // Control refresh timing so neither supplier learns that node 0 recovered before failover. + ScheduledThreadPoolExecutor executor = + new ScheduledThreadPoolExecutor(1) { + @Override + public ScheduledFuture scheduleAtFixedRate( + Runnable command, long initialDelay, long period, TimeUnit unit) { + 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()) { + business.run(); + background.run(); + assertEquals( + new HashSet<>(Arrays.asList(endpoint(1), endpoint(2))), new HashSet<>(business.get())); + assertEquals(business.get(), background.get()); + session.open(false, 1000, new ConcurrentHashMap<>(), business); + + EnvFactory.getEnv().startDataNode(0); + awaitStatus(0, NodeStatus.Running); + 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))); + + // Both discovered endpoints are dead. Only the initial seed omitted by discovery is live. + try (SessionDataSet result = session.executeQueryStatement("SHOW AVAILABLE URLS")) { + assertEquals(Collections.singleton(endpoint(0)), readEndpoints(result)); + } + background.run(); + assertEquals(Collections.singletonList(endpoint(0)), background.get()); + assertEquals(3, background.getSeedNodes().size()); + // 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()); + } finally { + business.close(); + background.close(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + private void assertQuery(Session session, SessionPool pool) throws Exception { + String sql = "SELECT s FROM root.seed_fallback.d"; + try (SessionDataSet result = session.executeQueryStatement(sql)) { + assertTrue(result.hasNext()); + assertEquals(42, result.next().getFields().get(0).getIntV()); + assertFalse(result.hasNext()); + } + try (SessionDataSetWrapper result = pool.executeQueryStatement(sql)) { + assertTrue(result.hasNext()); + assertEquals(42, result.next().getFields().get(0).getIntV()); + assertFalse(result.hasNext()); + } + } + + private NodesSupplier supplier(Object owner) throws Exception { + Field field = owner.getClass().getDeclaredField("availableNodes"); + field.setAccessible(true); + return (NodesSupplier) field.get(owner); + } + + private NodesSupplier createSupplier(ScheduledThreadPoolExecutor executor) { + List seeds = Arrays.asList(endpoint(0), endpoint(1), endpoint(2)); + return NodesSupplier.createNodeSupplier( + seeds, + executor, + "root", + "root", + ZoneId.of("UTC"), + SessionConfig.DEFAULT_INITIAL_BUFFER_CAPACITY, + SessionConfig.DEFAULT_MAX_FRAME_SIZE, + 1000, + false, + null, + null, + null, + null, + "TLS", + false, + "V_1_0"); + } + + private Set readEndpoints(SessionDataSet dataSet) throws Exception { + Set result = new HashSet<>(); + SessionDataSet.DataIterator iterator = dataSet.iterator(); + while (iterator.next()) { + result.add(new TEndPoint(iterator.getString("RpcAddress"), iterator.getInt("RpcPort"))); + } + return result; + } + + private TEndPoint endpoint(int index) { + return new TEndPoint( + EnvFactory.getEnv().getDataNodeWrapper(index).getIp(), + EnvFactory.getEnv().getDataNodeWrapper(index).getPort()); + } + + private List urls() { + return Arrays.asList(endpoint(0), endpoint(1), endpoint(2)).stream() + .map(UrlUtils::convertTEndPointIpv4AndIpv6Url) + .collect(Collectors.toCollection(ArrayList::new)); + } + + private void awaitStatus(int index, NodeStatus status) { + EnvFactory.getEnv() + .ensureNodeStatus( + Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(index)), + Collections.singletonList(status)); + } +} diff --git a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/INodeSupplier.java b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/INodeSupplier.java index 60a2a5adb4347..9c8d32f946120 100644 --- a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/INodeSupplier.java +++ b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/INodeSupplier.java @@ -21,6 +21,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import java.util.Collections; import java.util.List; import java.util.Optional; import java.util.function.Supplier; @@ -30,4 +31,14 @@ public interface INodeSupplier extends Supplier> { void close(); Optional getQueryEndPoint(); + + /** Initial addresses to try after all discovered endpoints have failed. */ + default List getSeedNodes() { + return Collections.emptyList(); + } + + /** Request a discovery refresh after reconnecting through a seed. */ + default void requestRefresh() { + // Suppliers without automatic discovery have nothing to refresh. + } } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java index 81d83450f72e6..3a52c861938f9 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java @@ -30,12 +30,17 @@ import java.time.ZoneId; import java.util.ArrayList; +import java.util.Collections; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; import java.util.Optional; -import java.util.concurrent.CopyOnWriteArrayList; +import java.util.Set; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; public class NodesSupplier implements INodeSupplier, Runnable { @@ -54,10 +59,13 @@ public class NodesSupplier implements INodeSupplier, Runnable { private static final int FETCH_SIZE = 10_000; - // availableNodes won't be updated frequently, so we use CopyOnWriteArrayList which is thread-safe - // and is optimized for scenarios of reading more and writing less + // Publish complete snapshots; discovery must never overwrite the initial recovery addresses. @SuppressWarnings("java:S3077") - private volatile List availableNodes = new CopyOnWriteArrayList<>(); + private volatile List availableNodes; + + private final List seedNodes; + private final ScheduledExecutorService executorService; + private final AtomicBoolean refreshRequested = new AtomicBoolean(); private final boolean useSSL; private final String trustStore; @@ -108,6 +116,7 @@ public static NodesSupplier createNodeSupplier( NodesSupplier nodesSupplier = new NodesSupplier( endPointList, + executorService, userName, password, zoneId, @@ -133,6 +142,7 @@ public static NodesSupplier createNodeSupplier( private NodesSupplier( List endPointList, + ScheduledExecutorService executorService, String userName, String password, ZoneId zoneId, @@ -147,7 +157,11 @@ private NodesSupplier( String sslProtocol, boolean enableRPCCompression, String version) { - this.availableNodes.addAll(new HashSet<>(endPointList)); + this.seedNodes = + Collections.unmodifiableList( + endPointList.stream().distinct().map(TEndPoint::deepCopy).collect(Collectors.toList())); + this.availableNodes = seedNodes; + this.executorService = executorService; this.userName = userName; this.password = password; this.useSSL = useSSL; @@ -172,24 +186,54 @@ public List get() { } @Override - public void run() { + public List getSeedNodes() { + return seedNodes; + } + + @Override + public void requestRefresh() { + if (!closed && refreshRequested.compareAndSet(false, true)) { + try { + executorService.execute( + () -> { + try { + run(); + } finally { + refreshRequested.set(false); + } + }); + } catch (RejectedExecutionException e) { + // The session or pool may have shut down its executor concurrently. + refreshRequested.set(false); + } + } + } + + @Override + public synchronized void run() { if (closed) { if (client != null) { destroyCurrentClient(); } return; } - if (client == null) { - for (TEndPoint endPoint : availableNodes) { - if (createConnection(endPoint)) { - break; - } + Set attempted = new HashSet<>(); + if (client != null) { + attempted.add(client.endPoint); + if (updateDataNodeList()) { + return; } + destroyCurrentClient(); } - // make sure the following code block can run thread-safely with close() method. - synchronized (this) { - if (client != null && !updateDataNodeList()) { + // A stale discovery snapshot can omit the only node that has finished a rolling restart. + Set candidates = new LinkedHashSet<>(availableNodes); + candidates.addAll(seedNodes); + for (TEndPoint endPoint : candidates) { + if (attempted.add(endPoint)) { + if (createConnection(endPoint) && updateDataNodeList()) { + return; + } destroyCurrentClient(); } } @@ -235,10 +279,11 @@ public void close() { @Override public Optional getQueryEndPoint() { - if (availableNodes == null || availableNodes.isEmpty()) { + List snapshot = availableNodes; + if (snapshot.isEmpty()) { return Optional.empty(); } else { - return Optional.of(policy.chooseOne(get())); + return Optional.of(policy.chooseOne(snapshot)); } } @@ -266,9 +311,7 @@ private void updateAvailableNodes(SessionDataSet sessionDataSet) throws Exceptio } } } - // replace the older ones. - if (!res.isEmpty()) { - availableNodes = res; - } + // An empty successful result is authoritative too; seeds remain available for recovery. + availableNodes = Collections.unmodifiableList(res); } } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java index a6deda174913d..b9e8fe3652ecb 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java @@ -125,6 +125,7 @@ public class Session implements ISession { new LinkedBlockingQueue<>(SessionConfig.DEFAULT_SESSION_EXECUTOR_TASK_NUM), ThreadUtils.createThreadFactory("SessionExecutor", true)); protected List nodeUrls; + private final List initialNodeUrls; protected String username; protected String password; protected boolean useEncryptedPassword; @@ -367,6 +368,8 @@ public Session( boolean enableRedirection, Version version) { this.defaultEndPoint = new TEndPoint(host, rpcPort); + this.initialNodeUrls = + Collections.singletonList(UrlUtils.convertTEndPointIpv4AndIpv6Url(defaultEndPoint)); this.username = username; this.password = password; this.fetchSize = fetchSize; @@ -434,8 +437,9 @@ public Session( if (nodeUrls.isEmpty()) { throw new IllegalArgumentException(SessionMessages.NODE_URLS_EMPTY); } - nodeUrls = shuffleNodeUrls(nodeUrls); + nodeUrls = shuffleNodeUrls(new ArrayList<>(nodeUrls)); this.nodeUrls = nodeUrls; + this.initialNodeUrls = Collections.unmodifiableList(new ArrayList<>(nodeUrls)); this.username = username; this.password = password; this.fetchSize = fetchSize; @@ -451,11 +455,13 @@ public Session(AbstractSessionBuilder builder) { if (builder.nodeUrls.isEmpty()) { throw new IllegalArgumentException(SessionMessages.NODE_URLS_EMPTY); } - builder.nodeUrls = shuffleNodeUrls(builder.nodeUrls); - this.nodeUrls = builder.nodeUrls; + this.nodeUrls = shuffleNodeUrls(new ArrayList<>(builder.nodeUrls)); + this.initialNodeUrls = Collections.unmodifiableList(new ArrayList<>(nodeUrls)); this.enableQueryRedirection = true; } else { this.defaultEndPoint = new TEndPoint(builder.host, builder.rpcPort); + this.initialNodeUrls = + Collections.singletonList(UrlUtils.convertTEndPointIpv4AndIpv6Url(defaultEndPoint)); this.enableQueryRedirection = builder.enableRedirection; } this.enableThriftRpcCompaction = builder.isThriftRpcCompactionEnabled; @@ -599,11 +605,7 @@ private static List shuffleNodeUrls(List endPoints) { } private List getNodeUrls() { - if (defaultEndPoint != null) { - return Collections.singletonList(defaultEndPoint); - } else { - return SessionUtils.parseSeedNodeUrls(nodeUrls); - } + return SessionUtils.parseSeedNodeUrls(initialNodeUrls); } @Override diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java index 7b6ec3b03dfea..1dde4a5c35ec7 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java @@ -22,6 +22,7 @@ import org.apache.iotdb.common.rpc.thrift.TAggregationType; import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; @@ -84,6 +85,7 @@ import java.time.ZoneId; import java.util.ArrayList; import java.util.Collections; +import java.util.LinkedHashSet; import java.util.List; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; @@ -1143,18 +1145,18 @@ private boolean reconnect() { for (int i = 1; i <= SessionConfig.RETRY_NUM; i++) { if (transport != null) { transport.close(); - endPointList = availableNodes.get(); - int currHostIndex = random.nextInt(endPointList.size()); - int tryHostNum = 0; - for (int j = currHostIndex; j < endPointList.size(); j++) { - if (tryHostNum == endPointList.size()) { - break; - } - this.endPoint = endPointList.get(j); - if (j == endPointList.size() - 1) { - j = -1; - } - tryHostNum++; + List discoveredNodes = availableNodes.get(); + endPointList = new ArrayList<>(discoveredNodes); + if (!endPointList.isEmpty()) { + Collections.rotate(endPointList, -random.nextInt(endPointList.size())); + } + if (availableNodes instanceof INodeSupplier) { + endPointList.addAll(((INodeSupplier) availableNodes).getSeedNodes()); + } + // Try every discovered endpoint before falling back to seeds, once per endpoint per round. + endPointList = new ArrayList<>(new LinkedHashSet<>(endPointList)); + for (TEndPoint candidate : endPointList) { + this.endPoint = candidate; try { init( endPoint, @@ -1165,6 +1167,9 @@ private boolean reconnect() { session.keyStorePwd, session.sslProtocol); connectedSuccess = true; + if (!discoveredNodes.contains(endPoint) && availableNodes instanceof INodeSupplier) { + ((INodeSupplier) availableNodes).requestRefresh(); + } } catch (IoTDBConnectionException e) { logger.warn(SessionMessages.NODE_DOWN_TRY_NEXT, endPoint); continue; diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java index 32feac2cea5e1..701979c47ac6b 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java @@ -52,6 +52,7 @@ import java.io.IOException; import java.time.ZoneId; +import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; @@ -564,7 +565,7 @@ public SessionPool(AbstractSessionPoolBuilder builder) { if (builder.nodeUrls.isEmpty()) { throw new IllegalArgumentException(SessionMessages.NODE_URLS_EMPTY); } - this.nodeUrls = builder.nodeUrls; + this.nodeUrls = Collections.unmodifiableList(new ArrayList<>(builder.nodeUrls)); this.host = null; this.port = -1; this.formattedNodeUrls = builder.nodeUrls.toString(); diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java new file mode 100644 index 0000000000000..bb1d3bfd5eec4 --- /dev/null +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java @@ -0,0 +1,411 @@ +/* + * 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.iotdb.session; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.isession.SessionConfig; +import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; +import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.UrlUtils; +import org.apache.iotdb.service.rpc.thrift.IClientRPCService; +import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp; +import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp; +import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion; + +import org.apache.thrift.TException; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.server.TThreadPoolServer; +import org.apache.thrift.transport.TServerSocket; +import org.apache.thrift.transport.TTransport; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.block.TsBlockBuilder; +import org.apache.tsfile.read.common.block.column.TsBlockSerde; +import org.apache.tsfile.utils.BytesUtils; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.powermock.reflect.Whitebox; + +import java.net.ServerSocket; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Exercise discovery and reconnection over real Thrift connections with controlled availability. + */ +public class SessionSeedNodeFallbackTest { + private final List nodes = new ArrayList<>(); + private final List attempts = new CopyOnWriteArrayList<>(); + private final List refreshes = new ArrayList<>(); + private final List connections = new ArrayList<>(); + private volatile List advertised; + private ScheduledExecutorService executor; + private NodesSupplier supplier; + + @Before + public void setUp() throws Exception { + for (int i = 0; i < 3; i++) { + nodes.add(new TestNode()); + } + advertised = endpoints(0, 1, 2); + executor = mock(ScheduledExecutorService.class); + doAnswer( + invocation -> { + refreshes.add(invocation.getArgument(0)); + return null; + }) + .when(executor) + .execute(any(Runnable.class)); + supplier = createSupplier(endpoints(0, 1, 2)); + } + + @After + public void tearDown() throws Exception { + for (SessionConnection connection : connections) { + connection.forceCloseTransport(); + } + if (supplier != null) { + supplier.close(); + } + for (TestNode node : nodes) { + node.close(); + } + } + + @Test + public void testBackgroundRefreshRecoversFromStaleSnapshot() { + nodes.get(0).online = false; + advertised = endpoints(1, 2); + supplier.run(); + assertEquals(advertised, supplier.get()); + + // Node 1 has finished upgrading, but both cached candidates are now unavailable. + nodes.get(0).online = true; + nodes.get(1).online = false; + nodes.get(2).online = false; + advertised = endpoints(0); + attempts.clear(); + supplier.run(); + assertEquals(endpoints(2, 0), attempts); + assertEquals(endpoints(0), supplier.get()); + assertEquals(endpoints(0, 1, 2), supplier.getSeedNodes()); + } + + @Test + public void testBusinessReconnectFallsBackAndRequestsRefresh() throws Exception { + advertised = endpoints(1, 2); + supplier.run(); + nodes.get(1).online = false; + nodes.get(2).online = false; + advertised = endpoints(0); + attempts.clear(); + SessionConnection connection = newConnection(); + assertTrue(reconnect(connection)); + assertEquals(nodes.get(0).endpoint, connection.getEndPoint()); + assertEquals(3, attempts.size()); + assertEquals(nodes.get(0).endpoint, attempts.get(2)); + assertEquals(1, refreshes.size()); + refreshes.remove(0).run(); + assertEquals(endpoints(0), supplier.get()); + } + + @Test + public void testDiscoveredNodeIsPreferredToHealthySeed() throws Exception { + advertised = endpoints(1); + supplier.run(); + attempts.clear(); + assertTrue(reconnect(newConnection())); + assertEquals(endpoints(1), attempts); + assertTrue(refreshes.isEmpty()); + for (int i = 0; i < 6; i++) { + assertEquals(nodes.get(1).endpoint, supplier.getQueryEndPoint().get()); + } + } + + @Test + public void testEmptyDiscoveryFallsBackWithoutLosingSeeds() throws Exception { + advertised = Collections.emptyList(); + supplier.run(); + assertTrue(supplier.get().isEmpty()); + assertFalse(supplier.getQueryEndPoint().isPresent()); + assertTrue(reconnect(newConnection())); + assertEquals(endpoints(0, 1, 2), supplier.getSeedNodes()); + advertised = endpoints(0, 1, 2); + refreshes.remove(0).run(); + assertEquals(advertised, supplier.get()); + } + + @Test + public void testAllUnavailableRetriesAreBoundedAndDeduplicated() throws Exception { + supplier.close(); + supplier = createSupplier(endpoints(0, 0, 1, 2)); + advertised = endpoints(1); + supplier.run(); + nodes.forEach(node -> node.online = false); + attempts.clear(); + assertFalse(reconnect(newConnection())); + assertEquals(SessionConfig.RETRY_NUM * 3, attempts.size()); + for (int i = 0; i < SessionConfig.RETRY_NUM; i++) { + assertEquals(endpoints(1, 0, 2), attempts.subList(i * 3, (i + 1) * 3)); + } + assertTrue(refreshes.isEmpty()); + } + + @Test + public void testSeedSnapshotIsIndependentOfCallerAndDiscovery() { + supplier.close(); + List seeds = endpoints(0, 1); + supplier = createSupplier(seeds); + seeds.get(0).setPort(1); + seeds.clear(); + advertised = endpoints(2); + supplier.run(); + assertEquals(endpoints(0, 1), supplier.getSeedNodes()); + assertEquals(endpoints(2), supplier.get()); + } + + @Test + public void testBackgroundRefreshDoesNotContactSeedsWhileCurrentNodeWorks() { + advertised = endpoints(1); + supplier.run(); + attempts.clear(); + supplier.run(); + assertTrue(attempts.isEmpty()); + assertEquals(endpoints(1), supplier.get()); + } + + @Test + public void testSessionRetainsInitialUrlsAfterDefaultEndpointChanges() throws Exception { + List urls = new ArrayList<>(Arrays.asList(url(0), url(1))); + Session session = new Session.Builder().nodeUrls(urls).build(); + urls.clear(); + session.defaultEndPoint = nodes.get(2).endpoint; + List seeds = Whitebox.invokeMethod(session, "getNodeUrls"); + assertEquals(2, seeds.size()); + assertTrue(seeds.containsAll(endpoints(0, 1))); + + Session singleNode = + new Session(nodes.get(0).endpoint.getIp(), nodes.get(0).endpoint.getPort()); + singleNode.defaultEndPoint = nodes.get(2).endpoint; + assertEquals(endpoints(0), Whitebox.invokeMethod(singleNode, "getNodeUrls")); + } + + @Test + public void testQuerySelectionDuringEmptyDiscoveryRefresh() throws Exception { + advertised = endpoints(0); + supplier.run(); + ExecutorService reader = Executors.newSingleThreadExecutor(); + AtomicBoolean finished = new AtomicBoolean(); + CountDownLatch started = new CountDownLatch(1); + Future selection = + reader.submit( + () -> { + started.countDown(); + while (!finished.get()) { + supplier + .getQueryEndPoint() + .ifPresent(endpoint -> assertEquals(nodes.get(0).endpoint, endpoint)); + } + }); + try { + assertTrue(started.await(5, TimeUnit.SECONDS)); + for (int i = 0; i < 100; i++) { + advertised = i % 2 == 0 ? Collections.emptyList() : endpoints(0); + supplier.run(); + } + } finally { + finished.set(true); + reader.shutdownNow(); + assertTrue(reader.awaitTermination(5, TimeUnit.SECONDS)); + } + selection.get(5, TimeUnit.SECONDS); + } + + @Test + public void testRefreshRequestsAreCoalescedAndIgnoredAfterClose() { + supplier.requestRefresh(); + supplier.requestRefresh(); + assertEquals(1, refreshes.size()); + refreshes.remove(0).run(); + supplier.requestRefresh(); + assertEquals(1, refreshes.size()); + supplier.close(); + refreshes.remove(0).run(); + supplier.requestRefresh(); + assertTrue(refreshes.isEmpty()); + } + + private NodesSupplier createSupplier(List seeds) { + return NodesSupplier.createNodeSupplier( + seeds, + executor, + "root", + "root", + ZoneId.of("UTC"), + SessionConfig.DEFAULT_INITIAL_BUFFER_CAPACITY, + SessionConfig.DEFAULT_MAX_FRAME_SIZE, + 1000, + false, + null, + null, + null, + null, + "TLS", + false, + "V_1_0"); + } + + private SessionConnection newConnection() { + SessionConnection connection = new SessionConnection("tree"); + Session session = new Session.Builder().nodeUrls(Collections.singletonList(url(0))).build(); + session.connectionTimeoutInMs = 1000; + Whitebox.setInternalState(connection, "session", session); + Whitebox.setInternalState(connection, "zoneId", ZoneId.of("UTC")); + Whitebox.setInternalState(connection, "transport", mock(TTransport.class)); + Whitebox.setInternalState(connection, "availableNodes", supplier); + connections.add(connection); + return connection; + } + + private boolean reconnect(SessionConnection connection) throws Exception { + return Whitebox.invokeMethod(connection, "reconnect"); + } + + private List endpoints(int... indices) { + List result = new ArrayList<>(); + for (int index : indices) { + result.add(nodes.get(index).endpoint.deepCopy()); + } + return result; + } + + private String url(int index) { + return UrlUtils.convertTEndPointIpv4AndIpv6Url(nodes.get(index).endpoint); + } + + private TSStatus success() { + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } + + private class TestNode implements AutoCloseable { + private final TEndPoint endpoint; + private final TServerSocket socket; + private final TThreadPoolServer server; + private final ExecutorService workers = Executors.newCachedThreadPool(); + private final Thread thread; + private volatile boolean online = true; + + private TestNode() throws Exception { + ServerSocket listener = new ServerSocket(0); + endpoint = new TEndPoint("127.0.0.1", listener.getLocalPort()); + socket = new TServerSocket(listener); + IClientRPCService.Iface handler = mock(IClientRPCService.Iface.class); + when(handler.openSession(any())) + .thenAnswer( + invocation -> { + attempts.add(endpoint); + checkOnline(); + return new TSOpenSessionResp(success(), TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3) + .setSessionId(1); + }); + when(handler.requestStatementId(anyLong())).thenReturn(1L); + when(handler.closeSession(any())).thenReturn(success()); + when(handler.closeOperation(any())).thenReturn(success()); + when(handler.executeQueryStatementV2(any())) + .thenAnswer( + invocation -> { + checkOnline(); + TsBlockBuilder block = + new TsBlockBuilder(Arrays.asList(TSDataType.TEXT, TSDataType.INT32)); + for (TEndPoint address : advertised) { + block.getTimeColumnBuilder().writeLong(0); + block.getColumnBuilder(0).writeBinary(BytesUtils.valueOf(address.getIp())); + block.getColumnBuilder(1).writeInt(address.getPort()); + block.declarePosition(); + } + return new TSExecuteStatementResp(success()) + .setQueryId(1) + .setColumns(Arrays.asList("RpcAddress", "RpcPort")) + .setDataTypeList(Arrays.asList("TEXT", "INT32")) + .setQueryResult( + advertised.isEmpty() + ? Collections.emptyList() + : Collections.singletonList( + new TsBlockSerde().serialize(block.build()))) + .setIgnoreTimeStamp(true) + .setMoreData(false); + }); + server = + new TThreadPoolServer( + new TThreadPoolServer.Args(socket) + .processor(new IClientRPCService.Processor<>(handler)) + .protocolFactory(new TBinaryProtocol.Factory()) + .transportFactory(DeepCopyRpcTransportFactory.INSTANCE) + .executorService(workers)); + CountDownLatch ready = new CountDownLatch(1); + thread = + new Thread( + () -> { + ready.countDown(); + server.serve(); + }, + "seed-fallback-test-server"); + thread.setDaemon(true); + thread.start(); + assertTrue(ready.await(5, TimeUnit.SECONDS)); + } + + private void checkOnline() throws TException { + if (!online) { + throw new TException("Node is restarting"); + } + } + + @Override + public void close() throws Exception { + server.stop(); + socket.close(); + workers.shutdownNow(); + thread.join(5000); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTask.java index 6d957769b4428..9c1311a77fe0a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTask.java @@ -19,7 +19,7 @@ package org.apache.iotdb.db.queryengine.plan.execution.config.metadata; -import org.apache.iotdb.commons.cluster.RegionStatus; +import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.schema.column.ColumnHeader; import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo; @@ -58,7 +58,9 @@ public static void buildTsBlock( if (showDataNodesResp.getDataNodesInfoList() != null) { for (TDataNodeInfo dataNodeInfo : showDataNodesResp.getDataNodesInfoList()) { String status = dataNodeInfo.getStatus(); - if (RegionStatus.Removing.getStatus().equals(status)) { + if (NodeStatus.Removing.getStatus().equals(status) + || NodeStatus.Unknown.getStatus().equals(status) + || NodeStatus.Stopped.getStatus().equals(status)) { continue; } builder.getTimeColumnBuilder().writeLong(0L); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTaskTest.java new file mode 100644 index 0000000000000..734915bcff9d4 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTaskTest.java @@ -0,0 +1,78 @@ +/* + * 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.iotdb.db.queryengine.plan.execution.config.metadata; + +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo; +import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp; +import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult; + +import com.google.common.util.concurrent.SettableFuture; +import org.apache.tsfile.read.common.block.TsBlock; +import org.junit.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +import static org.junit.Assert.assertEquals; + +public class ShowAvailableUrlsTaskTest { + @Test + public void testOnlyRunningAndReadOnlyNodesAreReturned() throws Exception { + List nodes = + Arrays.stream(NodeStatus.values()) + .map( + status -> + new TDataNodeInfo() + .setStatus(status.getStatus()) + .setStatusReason("reason") + .setRpcAddresss("127.0.0.1") + .setRpcPort(6667 + status.ordinal())) + .collect(Collectors.toList()); + TsBlock result = execute(nodes); + assertEquals(2, result.getPositionCount()); + assertEquals("127.0.0.1", result.getColumn(0).getBinary(0).toString()); + assertEquals(6667 + NodeStatus.Running.ordinal(), result.getColumn(1).getInt(0)); + assertEquals(6667 + NodeStatus.ReadOnly.ordinal(), result.getColumn(1).getInt(1)); + } + + @Test + public void testNoOnlineNodes() throws Exception { + assertEquals( + 0, + execute( + Arrays.asList( + new TDataNodeInfo().setStatus(NodeStatus.Unknown.getStatus()), + new TDataNodeInfo().setStatus(NodeStatus.Stopped.getStatus()), + new TDataNodeInfo().setStatus(NodeStatus.Removing.getStatus()))) + .getPositionCount()); + assertEquals(0, execute(Collections.emptyList()).getPositionCount()); + assertEquals(0, execute(null).getPositionCount()); + } + + private TsBlock execute(List nodes) throws Exception { + SettableFuture future = SettableFuture.create(); + ShowAvailableUrlsTask.buildTsBlock( + new TShowDataNodesResp().setDataNodesInfoList(nodes), future); + return future.get().getResultSet(); + } +} From c24019780a34b2f5f66e6279ef702ae692bfb0b4 Mon Sep 17 00:00:00 2001 From: JackieTien97 Date: Tue, 20 Oct 2026 12:07:50 +0800 Subject: [PATCH 2/3] Fix node discovery lifecycle and recovery edge cases --- RELEASE_NOTES.md | 26 ++ .../apache/iotdb/session/NodesSupplier.java | 221 +++++++++++--- .../org/apache/iotdb/session/Session.java | 9 +- .../iotdb/session/SessionConnection.java | 6 +- .../iotdb/session/TableSessionBuilder.java | 3 + .../iotdb/session/ThriftConnection.java | 127 +++++--- .../iotdb/session/pool/SessionPool.java | 10 +- .../session/pool/TableSessionPoolBuilder.java | 3 + .../session/SessionSeedNodeFallbackTest.java | 288 ++++++++++++++++-- .../ThriftConnectionLifecycleTest.java | 207 +++++++++++++ .../metadata/ShowAvailableUrlsTask.java | 4 +- .../metadata/ShowAvailableUrlsTaskTest.java | 23 ++ 12 files changed, 823 insertions(+), 104 deletions(-) create mode 100644 iotdb-client/session/src/test/java/org/apache/iotdb/session/ThriftConnectionLifecycleTest.java diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 96b941b107d18..851a79f429b9f 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -19,6 +19,32 @@ --> +# Unreleased + +## Java client discovery and rolling upgrades + +`SHOW AVAILABLE URLS` excludes `Unknown`, `Stopped`, and `Removing` DataNodes, as well as +`ReadOnly` nodes whose status reason is `Stopping`. Other `ReadOnly` nodes remain query endpoints. +The Java SDK keeps its initial `nodeUrls` as recovery seeds when the latest discovered endpoints +cannot be reached. Configure several stable seeds and keep at least one reachable during a rolling +upgrade. + +Upgrade Java clients to a version with seed fallback **before upgrading the servers**. The server +filter applies to every client. Older Java SDKs and the C++ client do not gain seed fallback from a +server upgrade: their discovered list may lose a node that later becomes the only reachable node. +For those clients, keep an endpoint in their last discovery snapshot reachable throughout maintenance, +or use a stable explicit endpoint list with automatic discovery disabled until the client supports +seed fallback. This change does not port the Java recovery mechanism to C++. + +Seed and discovered endpoints are deduplicated by their literal host/IP and port, without DNS +resolution. Configure `nodeUrls` using the same address form as each DataNode's advertised +`dn_rpc_address`; for example, do not mix `localhost` seeds with `127.0.0.1` discovery addresses for +the same node. For wildcard bind addresses such as `0.0.0.0`, supply reachable addresses in +`nodeUrls`; wildcard discovery rows do not clear the previous snapshot. Aliases can otherwise cause +duplicate connection attempts, extra discovery refreshes, +and reconnections on retryable responses. DNS names may represent multiple or changing addresses, +so recovery does not resolve them to infer permanent node identity. + # Apache IoTDB 2.0.10 ## Features & Improvements diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java index 3a52c861938f9..572050edeaf3e 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java @@ -22,9 +22,11 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.UrlUtils; import org.apache.iotdb.session.i18n.SessionMessages; +import org.apache.thrift.TApplicationException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,8 +40,10 @@ import java.util.Set; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; public class NodesSupplier implements INodeSupplier, Runnable { @@ -90,13 +94,57 @@ public class NodesSupplier implements INodeSupplier, Runnable { private final QueryEndPointPolicy policy = new RoundRobinPolicy(); + // Refresh serialization is independent of the short client-publication/cleanup critical section. + private final ReentrantLock refreshLock = new ReentrantLock(); + private final Object clientLock = new Object(); private ThriftConnection client; + private ThriftConnection activeClient; private volatile boolean closed = false; + public static NodesSupplier createNodeSupplier( + List endPointList, + ScheduledExecutorService executorService, + String userName, + String password, + ZoneId zoneId, + int thriftDefaultBufferSize, + int thriftMaxFrameSize, + int connectionTimeoutInMs, + boolean useSSL, + String trustStore, + String trustStorePwd, + String keyStore, + String keyStorePwd, + String sslProtocol, + boolean enableRPCCompression, + String version) { + + return createNodeSupplier( + endPointList, + endPointList, + executorService, + userName, + password, + zoneId, + thriftDefaultBufferSize, + thriftMaxFrameSize, + connectionTimeoutInMs, + useSSL, + trustStore, + trustStorePwd, + keyStore, + keyStorePwd, + sslProtocol, + enableRPCCompression, + version); + } + + /** Bootstrap discovery independently of the permanent recovery seeds (for Session reopen). */ @SuppressWarnings("unsafeThreadSchedule") public static NodesSupplier createNodeSupplier( List endPointList, + List seedNodes, ScheduledExecutorService executorService, String userName, String password, @@ -116,6 +164,7 @@ public static NodesSupplier createNodeSupplier( NodesSupplier nodesSupplier = new NodesSupplier( endPointList, + seedNodes, executorService, userName, password, @@ -142,6 +191,7 @@ public static NodesSupplier createNodeSupplier( private NodesSupplier( List endPointList, + List seedNodes, ScheduledExecutorService executorService, String userName, String password, @@ -158,9 +208,11 @@ private NodesSupplier( boolean enableRPCCompression, String version) { this.seedNodes = + Collections.unmodifiableList( + seedNodes.stream().distinct().map(TEndPoint::deepCopy).collect(Collectors.toList())); + this.availableNodes = Collections.unmodifiableList( endPointList.stream().distinct().map(TEndPoint::deepCopy).collect(Collectors.toList())); - this.availableNodes = seedNodes; this.executorService = executorService; this.userName = userName; this.password = password; @@ -178,8 +230,7 @@ private NodesSupplier( this.version = version; } - // the method will only be called while reconnecting, so it's ok to copy set each time - // and the List needn't be thread-safe, because it will only be used in one thread. + // Readers retain an immutable snapshot while discovery publishes replacements. @Override public List get() { return availableNodes; @@ -210,41 +261,91 @@ public void requestRefresh() { } @Override - public synchronized void run() { - if (closed) { - if (client != null) { - destroyCurrentClient(); - } + public void run() { + if (closed || !refreshLock.tryLock()) { return; } - Set attempted = new HashSet<>(); - if (client != null) { - attempted.add(client.endPoint); - if (updateDataNodeList()) { - return; + try { + Set attempted = new HashSet<>(); + ThriftConnection current; + synchronized (clientLock) { + if (closed) { + return; + } + current = client; + activeClient = current; + } + if (current != null) { + attempted.add(current.endPoint); + if (refreshFrom(current)) { + return; + } } - destroyCurrentClient(); - } - // A stale discovery snapshot can omit the only node that has finished a rolling restart. - Set candidates = new LinkedHashSet<>(availableNodes); - candidates.addAll(seedNodes); - for (TEndPoint endPoint : candidates) { - if (attempted.add(endPoint)) { - if (createConnection(endPoint) && updateDataNodeList()) { + Set candidates = new LinkedHashSet<>(availableNodes); + List seeds = new ArrayList<>(seedNodes); + if (!seeds.isEmpty()) { + Collections.rotate(seeds, -ThreadLocalRandom.current().nextInt(seeds.size())); + } + candidates.addAll(seeds); + for (TEndPoint endPoint : candidates) { + if (closed) { return; } - destroyCurrentClient(); + if (attempted.add(endPoint)) { + current = createConnection(endPoint); + if (current != null && refreshFrom(current)) { + return; + } + } } + } finally { + ThriftConnection toClose = null; + synchronized (clientLock) { + if (closed) { + toClose = activeClient; + client = null; + } + activeClient = null; + } + // Only the refresh owner releases buffers that an interrupted operation could still use. + try { + if (toClose != null) { + toClose.close(); + } + } finally { + refreshLock.unlock(); + } + } + } + + /** Return false only when the connection failed and another candidate should be tried. */ + private boolean refreshFrom(ThriftConnection current) { + if (closed) { + return true; + } + try { + updateDataNodeList(current); + } catch (IoTDBConnectionException e) { + destroyClient(current); + return false; } + return true; } - private boolean createConnection(TEndPoint endPoint) { - client = + private ThriftConnection createConnection(TEndPoint endPoint) { + ThriftConnection current = new ThriftConnection( endPoint, thriftDefaultBufferSize, thriftMaxFrameSize, connectionTimeoutInMs); + synchronized (clientLock) { + if (closed) { + return null; + } + client = current; + activeClient = current; + } try { - client.init( + current.init( useSSL, trustStore, trustStorePwd, @@ -256,25 +357,46 @@ private boolean createConnection(TEndPoint endPoint) { enableRPCCompression, zoneId, version); - return true; + if (closed) { + destroyClient(current); + return null; + } + return current; } catch (Exception e) { - LOGGER.warn(SessionMessages.FAILED_TO_CREATE_CONNECTION, endPoint); - destroyCurrentClient(); - return false; + if (!closed) { + LOGGER.warn(SessionMessages.FAILED_TO_CREATE_CONNECTION, endPoint); + } + destroyClient(current); + return null; } } - private synchronized void destroyCurrentClient() { - if (client != null) { - client.close(); - client = null; + private void destroyClient(ThriftConnection current) { + synchronized (clientLock) { + if (client == current) { + client = null; + } } + current.close(); } @Override public void close() { - closed = true; - destroyCurrentClient(); + ThriftConnection current; + boolean inUse; + synchronized (clientLock) { + closed = true; + current = client != null ? client : activeClient; + inUse = current != null && current == activeClient; + client = null; + } + if (current != null) { + if (inUse) { + current.abort(); + } else { + current.close(); + } + } } @Override @@ -287,15 +409,25 @@ public Optional getQueryEndPoint() { } } - private boolean updateDataNodeList() { + private void updateDataNodeList(ThriftConnection current) throws IoTDBConnectionException { try (SessionDataSet sessionDataSet = - client.executeQueryStatement(SHOW_AVAILABLE_URLS_COMMAND, TIMEOUT_IN_MS, FETCH_SIZE)) { + current.executeQueryStatement(SHOW_AVAILABLE_URLS_COMMAND, TIMEOUT_IN_MS, FETCH_SIZE)) { updateAvailableNodes(sessionDataSet); - } catch (Exception e1) { - LOGGER.warn(SessionMessages.FAILED_TO_FETCH_DATA_NODE_LIST, client.endPoint); - return false; + } catch (IoTDBConnectionException e) { + if (!closed) { + LOGGER.warn(SessionMessages.FAILED_TO_FETCH_DATA_NODE_LIST, current.endPoint); + } + if (!(e.getCause() instanceof TApplicationException)) { + throw e; + } + // A Thrift application error also means the peer responded; it is not a transport failure. + } catch (Exception e) { + // A query error (for example no ConfigNode leader) is not evidence of a dead endpoint. + // Keep this connection and snapshot, and let the next scheduled refresh retry the query. + if (!closed) { + LOGGER.warn(SessionMessages.FAILED_TO_FETCH_DATA_NODE_LIST, current.endPoint); + } } - return true; } private void updateAvailableNodes(SessionDataSet sessionDataSet) throws Exception { @@ -311,7 +443,10 @@ private void updateAvailableNodes(SessionDataSet sessionDataSet) throws Exceptio } } } - // An empty successful result is authoritative too; seeds remain available for recovery. - availableNodes = Collections.unmodifiableList(res); + // Wildcard addresses and transient Unknown statuses after leader changes convey no usable + // discovery information. Retain the last snapshot for routing and recovery in both cases. + if (!closed && !res.isEmpty()) { + availableNodes = Collections.unmodifiableList(res); + } } } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java index b9e8fe3652ecb..9ac9c9d8cb9d5 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java @@ -546,6 +546,7 @@ public synchronized void open(boolean enableRPCCompaction, int connectionTimeout this.availableNodes = NodesSupplier.createNodeSupplier( getNodeUrls(), + SessionUtils.parseSeedNodeUrls(initialNodeUrls), executorService, username, password, @@ -605,7 +606,9 @@ private static List shuffleNodeUrls(List endPoints) { } private List getNodeUrls() { - return SessionUtils.parseSeedNodeUrls(initialNodeUrls); + return defaultEndPoint == null + ? SessionUtils.parseSeedNodeUrls(initialNodeUrls) + : Collections.singletonList(defaultEndPoint.deepCopy()); } @Override @@ -4493,6 +4496,10 @@ public Builder enableRecordsAutoConvertTablet(boolean enableRecordsAutoConvertTa return this; } + /** + * Configure recovery seeds using the same host/IP form as each DataNode's dn_rpc_address. + * Hostname aliases and IP literals are not DNS-deduplicated during recovery. + */ public Builder nodeUrls(List nodeUrls) { this.nodeUrls = nodeUrls; return this; diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java index 1dde4a5c35ec7..ce06b35abb106 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java @@ -1151,7 +1151,11 @@ private boolean reconnect() { Collections.rotate(endPointList, -random.nextInt(endPointList.size())); } if (availableNodes instanceof INodeSupplier) { - endPointList.addAll(((INodeSupplier) availableNodes).getSeedNodes()); + List seeds = new ArrayList<>(((INodeSupplier) availableNodes).getSeedNodes()); + if (!seeds.isEmpty()) { + Collections.rotate(seeds, -random.nextInt(seeds.size())); + } + endPointList.addAll(seeds); } // Try every discovered endpoint before falling back to seeds, once per endpoint per round. endPointList = new ArrayList<>(new LinkedHashSet<>(endPointList)); diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java index 20147e1cfcf29..a39af4c847a95 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java @@ -49,6 +49,9 @@ public class TableSessionBuilder extends AbstractSessionBuilder { /** * Sets the list of node URLs for the IoTDB cluster. * + *

Use the same host/IP form as each DataNode's dn_rpc_address. Recovery deduplicates literal + * addresses, not DNS aliases. + * * @param nodeUrls a list of node URLs. * @return the current {@link TableSessionBuilder} instance. * @defaultValue Collection.singletonList(" localhost : 6667 ") diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/ThriftConnection.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/ThriftConnection.java index 4f8e2a58af758..71cc63526d711 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/ThriftConnection.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/ThriftConnection.java @@ -25,8 +25,8 @@ import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.rpc.TElasticFramedTransport; import org.apache.iotdb.service.rpc.thrift.IClientRPCService; -import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq; import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq; import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp; import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq; @@ -36,12 +36,16 @@ import org.apache.thrift.TException; import org.apache.thrift.protocol.TBinaryProtocol; import org.apache.thrift.protocol.TCompactProtocol; +import org.apache.thrift.transport.TSocket; import org.apache.thrift.transport.TTransport; import org.apache.thrift.transport.TTransportException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.IOException; +import java.net.Socket; import java.time.ZoneId; +import java.util.concurrent.atomic.AtomicBoolean; public class ThriftConnection { @@ -55,7 +59,10 @@ public class ThriftConnection { protected final int connectionTimeoutInMs; - protected TTransport transport; + protected volatile TTransport transport; + private final Object transportLock = new Object(); + private final AtomicBoolean disposed = new AtomicBoolean(); + private volatile boolean closed; protected IClientRPCService.Iface client; protected long sessionId; protected long statementId; @@ -87,31 +94,28 @@ public void init( ZoneId zoneId, String version) throws IoTDBConnectionException { - DeepCopyRpcTransportFactory transportFactory = - DeepCopyRpcTransportFactory.getInstance(thriftDefaultBufferSize, thriftMaxFrameSize); + checkNotClosed(); try { - if (useSSL) { - transport = - transportFactory.getTransport( - endPoint.getIp(), - endPoint.getPort(), - connectionTimeoutInMs, - trustStore, - trustStorePwd, - keyStore, - keyStorePwd, - sslProtocol); - } else { - transport = - transportFactory.getTransport( - // as there is a try-catch already, we do not need to use TSocket.wrap - endPoint.getIp(), endPoint.getPort(), connectionTimeoutInMs); + TTransport prepared = + createTransport(useSSL, trustStore, trustStorePwd, keyStore, keyStorePwd, sslProtocol); + boolean published; + synchronized (transportLock) { + published = !closed; + if (published) { + transport = prepared; + } } - if (!transport.isOpen()) { - transport.open(); + if (!published) { + prepared.close(); + throw new IoTDBConnectionException(SessionMessages.SESSION_NOT_OPEN); } + if (!prepared.isOpen()) { + prepared.open(); + } + checkNotClosed(); this.zoneId = zoneId; } catch (TTransportException e) { + close(); throw new IoTDBConnectionException(e); } @@ -155,7 +159,7 @@ public void init( statementId = client.requestStatementId(sessionId); } catch (Exception e) { - transport.close(); + close(); throw new IoTDBConnectionException(e); } } @@ -193,19 +197,74 @@ protected SessionDataSet executeQueryStatement(String sql, long timeout, int fet execResp.getColumnIndex2TsBlockColumnIndexList()); } - public void close() { - if (transport != null && transport.isOpen()) { - try { - if (client != null) { - client.closeSession(new TSCloseSessionReq(sessionId)); - } - } catch (TException e) { - LOGGER.warn(SessionMessages.CLOSING_SESSION_FAILED, sessionId, endPoint); - } finally { - if (transport.isOpen()) { - transport.close(); + /** Create outside the publication lock: TLS setup itself can perform network I/O. */ + protected TTransport createTransport( + boolean useSSL, + String trustStore, + String trustStorePwd, + String keyStore, + String keyStorePwd, + String sslProtocol) + throws TTransportException { + DeepCopyRpcTransportFactory factory = + DeepCopyRpcTransportFactory.getInstance(thriftDefaultBufferSize, thriftMaxFrameSize); + return useSSL + ? factory.getTransport( + endPoint.getIp(), + endPoint.getPort(), + connectionTimeoutInMs, + trustStore, + trustStorePwd, + keyStore, + keyStorePwd, + sslProtocol) + : factory.getTransport(endPoint.getIp(), endPoint.getPort(), connectionTimeoutInMs); + } + + private void checkNotClosed() throws IoTDBConnectionException { + if (closed) { + throw new IoTDBConnectionException(SessionMessages.SESSION_NOT_OPEN); + } + } + + /** + * Abort socket I/O without waiting for the synchronized RPC client or releasing buffers still + * owned by the refresh thread. Closing the Socket directly also prevents TSocket.open() from + * creating a replacement socket if closure races with the start of TCP connect. + */ + void abort() { + TTransport current; + synchronized (transportLock) { + closed = true; + current = transport; + } + if (current == null) { + return; + } + TTransport underlying = + current instanceof TElasticFramedTransport + ? ((TElasticFramedTransport) current).getSocket() + : current; + if (underlying instanceof TSocket) { + Socket socket = ((TSocket) underlying).getSocket(); + if (socket != null) { + try { + socket.close(); + } catch (IOException e) { + LOGGER.warn(SessionMessages.CLOSING_SESSION_FAILED, sessionId, endPoint, e); } } + } else { + underlying.close(); + } + } + + /** Release the transport after its refresh owner has finished using the buffers. */ + public void close() { + abort(); + if (disposed.compareAndSet(false, true) && transport != null) { + // Closing the transport also releases the server-side session through handleClientExit(). + transport.close(); } } } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java index 701979c47ac6b..16b79c6415c4b 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java @@ -503,7 +503,7 @@ public SessionPool( if (nodeUrls.isEmpty()) { throw new IllegalArgumentException(SessionMessages.NODE_URLS_EMPTY); } - this.nodeUrls = nodeUrls; + this.nodeUrls = Collections.unmodifiableList(new ArrayList<>(nodeUrls)); this.user = user; this.password = password; this.fetchSize = fetchSize; @@ -520,9 +520,9 @@ public SessionPool( this.version = version; this.thriftDefaultBufferSize = thriftDefaultBufferSize; this.thriftMaxFrameSize = thriftMaxFrameSize; - this.formattedNodeUrls = nodeUrls.toString(); + this.formattedNodeUrls = this.nodeUrls.toString(); initThreadPool(); - initAvailableNodes(SessionUtils.parseSeedNodeUrls(nodeUrls)); + initAvailableNodes(SessionUtils.parseSeedNodeUrls(this.nodeUrls)); } public SessionPool(AbstractSessionPoolBuilder builder) { @@ -3698,6 +3698,10 @@ public Builder port(int port) { return this; } + /** + * Configure recovery seeds using the same host/IP form as each DataNode's dn_rpc_address. + * Hostname aliases and IP literals are not DNS-deduplicated during recovery. + */ public Builder nodeUrls(List nodeUrls) { this.nodeUrls = nodeUrls; return this; diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java index 790e781628899..f19c5fc8878a6 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java @@ -39,6 +39,9 @@ public class TableSessionPoolBuilder extends AbstractSessionPoolBuilder { /** * Sets the list of node URLs for the IoTDB cluster. * + *

Use the same host/IP form as each DataNode's dn_rpc_address. Recovery deduplicates literal + * addresses, not DNS aliases. + * * @param nodeUrls a list of node URLs. * @return the current {@link TableSessionPoolBuilder} instance. * @defaultValue Collection.singletonList("localhost:6667") diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java index bb1d3bfd5eec4..b30d6a9a3367d 100644 --- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java @@ -22,6 +22,8 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.isession.SessionConfig; +import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.isession.pool.SessionDataSetWrapper; import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.rpc.UrlUtils; @@ -29,9 +31,14 @@ import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp; import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp; import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion; +import org.apache.iotdb.session.pool.SessionPool; +import org.apache.thrift.TApplicationException; import org.apache.thrift.TException; import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.server.ServerContext; +import org.apache.thrift.server.TServerEventHandler; import org.apache.thrift.server.TThreadPoolServer; import org.apache.thrift.transport.TServerSocket; import org.apache.thrift.transport.TTransport; @@ -45,10 +52,12 @@ import org.powermock.reflect.Whitebox; import java.net.ServerSocket; +import java.net.Socket; import java.time.ZoneId; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; @@ -61,6 +70,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; @@ -74,6 +84,7 @@ public class SessionSeedNodeFallbackTest { private final List nodes = new ArrayList<>(); private final List attempts = new CopyOnWriteArrayList<>(); + private final List queries = new CopyOnWriteArrayList<>(); private final List refreshes = new ArrayList<>(); private final List connections = new ArrayList<>(); private volatile List advertised; @@ -161,16 +172,76 @@ public void testDiscoveredNodeIsPreferredToHealthySeed() throws Exception { } @Test - public void testEmptyDiscoveryFallsBackWithoutLosingSeeds() throws Exception { + public void testEmptyDiscoveryKeepsPreviousSnapshot() { + advertised = endpoints(1, 2); + supplier.run(); + attempts.clear(); advertised = Collections.emptyList(); supplier.run(); - assertTrue(supplier.get().isEmpty()); - assertFalse(supplier.getQueryEndPoint().isPresent()); - assertTrue(reconnect(newConnection())); + assertEquals(endpoints(1, 2), supplier.get()); + assertEquals(nodes.get(1).endpoint, supplier.getQueryEndPoint().get()); + assertEquals(nodes.get(2).endpoint, supplier.getQueryEndPoint().get()); assertEquals(endpoints(0, 1, 2), supplier.getSeedNodes()); - advertised = endpoints(0, 1, 2); - refreshes.remove(0).run(); - assertEquals(advertised, supplier.get()); + assertTrue(attempts.isEmpty()); + } + + @Test + public void testWildcardDiscoveryKeepsSeedsAndLastSnapshot() { + advertised = Arrays.asList(new TEndPoint("0.0.0.0", 6667), new TEndPoint("::", 6667)); + supplier.run(); + assertEquals(endpoints(0, 1, 2), supplier.get()); + List selected = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + selected.add(supplier.getQueryEndPoint().get()); + } + assertEquals(new HashSet<>(endpoints(0, 1, 2)), new HashSet<>(selected)); + + advertised = endpoints(1, 2); + supplier.run(); + advertised = Collections.singletonList(new TEndPoint("0.0.0.0", 6667)); + supplier.run(); + assertEquals(endpoints(1, 2), supplier.get()); + } + + @Test + public void testQueryStatusFailureDoesNotRotateConnections() { + assertQueryFailureDoesNotRotate(false); + } + + @Test + public void testQueryApplicationFailureDoesNotRotateConnections() { + assertQueryFailureDoesNotRotate(true); + } + + private void assertQueryFailureDoesNotRotate(boolean applicationError) { + advertised = endpoints(1, 2); + supplier.run(); + nodes.forEach( + node -> { + node.failQuery = true; + node.queryApplicationError = applicationError; + }); + queries.clear(); + attempts.clear(); + supplier.run(); + assertEquals(endpoints(0), queries); + assertTrue(attempts.isEmpty()); + assertEquals(endpoints(1, 2), supplier.get()); + + nodes.forEach(node -> node.failQuery = false); + advertised = endpoints(2); + supplier.run(); + assertEquals(endpoints(2), supplier.get()); + assertTrue(attempts.isEmpty()); + } + + @Test + public void testQueryFailureAfterConnectingDoesNotTryOtherCandidates() { + nodes.forEach(node -> node.failQuery = true); + supplier.run(); + assertEquals(endpoints(0), attempts); + assertEquals(endpoints(0), queries); + assertEquals(endpoints(0, 1, 2), supplier.get()); } @Test @@ -184,9 +255,12 @@ public void testAllUnavailableRetriesAreBoundedAndDeduplicated() throws Exceptio assertFalse(reconnect(newConnection())); assertEquals(SessionConfig.RETRY_NUM * 3, attempts.size()); for (int i = 0; i < SessionConfig.RETRY_NUM; i++) { - assertEquals(endpoints(1, 0, 2), attempts.subList(i * 3, (i + 1) * 3)); + List round = attempts.subList(i * 3, (i + 1) * 3); + assertEquals(nodes.get(1).endpoint, round.get(0)); + assertEquals(new HashSet<>(endpoints(0, 1, 2)), new HashSet<>(round)); } assertTrue(refreshes.isEmpty()); + assertEquals(endpoints(0, 1, 2), supplier.getSeedNodes()); } @Test @@ -213,19 +287,141 @@ public void testBackgroundRefreshDoesNotContactSeedsWhileCurrentNodeWorks() { } @Test - public void testSessionRetainsInitialUrlsAfterDefaultEndpointChanges() throws Exception { - List urls = new ArrayList<>(Arrays.asList(url(0), url(1))); - Session session = new Session.Builder().nodeUrls(urls).build(); + public void testNodeUrlsSessionReopenAfterFailover() throws Exception { + assertSessionReopenAfterFailover(false); + } + + @Test + public void testSingleHostSessionReopenAfterFailover() throws Exception { + assertSessionReopenAfterFailover(true); + } + + private void assertSessionReopenAfterFailover(boolean singleHost) throws Exception { + List urls = new ArrayList<>(Collections.singletonList(url(0))); + Session session = + singleHost + ? new Session.Builder() + .host(nodes.get(0).endpoint.getIp()) + .port(nodes.get(0).endpoint.getPort()) + .build() + : new Session.Builder().nodeUrls(urls).build(); urls.clear(); - session.defaultEndPoint = nodes.get(2).endpoint; - List seeds = Whitebox.invokeMethod(session, "getNodeUrls"); - assertEquals(2, seeds.size()); - assertTrue(seeds.containsAll(endpoints(0, 1))); - - Session singleNode = - new Session(nodes.get(0).endpoint.getIp(), nodes.get(0).endpoint.getPort()); - singleNode.defaultEndPoint = nodes.get(2).endpoint; - assertEquals(endpoints(0), Whitebox.invokeMethod(singleNode, "getNodeUrls")); + advertised = endpoints(0, 1); + try { + session.open(false, 1000); + NodesSupplier first = Whitebox.getInternalState(session, "availableNodes"); + awaitSnapshot(first, endpoints(0, 1)); + nodes.get(0).online = false; + advertised = endpoints(1); + awaitSnapshot(first, endpoints(1)); + SessionConnection connection = Whitebox.getInternalState(session, "defaultSessionConnection"); + assertTrue(reconnect(connection)); + assertEquals(nodes.get(1).endpoint, connection.getEndPoint()); + session.close(); + + attempts.clear(); + session.open(false, 1000); + NodesSupplier reopened = Whitebox.getInternalState(session, "availableNodes"); + assertEquals(endpoints(0), reopened.getSeedNodes()); + assertEquals(endpoints(1), reopened.get()); + awaitSnapshot(reopened, endpoints(1)); + for (int i = 0; i < 10; i++) { + try (SessionDataSet result = session.executeQueryStatement("SHOW AVAILABLE URLS")) { + assertTrue(result.hasNext()); + result.next(); + assertFalse(result.hasNext()); + } + } + assertFalse(attempts.contains(nodes.get(0).endpoint)); + } finally { + session.close(); + } + } + + private void awaitSnapshot(NodesSupplier current, List expected) throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (!expected.equals(current.get()) && System.nanoTime() < deadline) { + current.run(); + TimeUnit.MILLISECONDS.sleep(10); + } + assertEquals(expected, current.get()); + } + + @Test + public void testPoolListConstructorCopiesUrlsForFutureSessions() throws Exception { + List urls = new ArrayList<>(Collections.singletonList(url(0))); + SessionPool pool = new SessionPool(urls, "root", "root", 1); + try { + urls.clear(); + try (SessionDataSetWrapper result = pool.executeQueryStatement("SHOW AVAILABLE URLS")) { + assertTrue(result.hasNext()); + } + NodesSupplier current = Whitebox.getInternalState(pool, "availableNodes"); + assertEquals(endpoints(0), current.getSeedNodes()); + } finally { + pool.close(); + } + } + + @Test + public void testCloseAbortsBlockedSessionOpenAndDoesNotTryRemainingCandidates() throws Exception { + supplier.close(); + ExecutorService tasks = Executors.newCachedThreadPool(); + Socket peer = null; + try (ServerSocket listener = new ServerSocket(0)) { + List seeds = new ArrayList<>(); + seeds.add(new TEndPoint("127.0.0.1", listener.getLocalPort())); + seeds.addAll(endpoints(1, 2)); + supplier = createSupplier(seeds, 0); + Future accepted = tasks.submit(listener::accept); + Future refresh = tasks.submit(supplier::run); + peer = accepted.get(5, TimeUnit.SECONDS); + peer.setSoTimeout(5000); + assertTrue(peer.getInputStream().read() >= 0); + // Timeout 0 means the peer can stall openSession forever unless close aborts the socket. + tasks.submit(supplier::close).get(5, TimeUnit.SECONDS); + refresh.get(5, TimeUnit.SECONDS); + assertTrue(attempts.isEmpty()); + assertNull(Whitebox.getInternalState(supplier, "client")); + assertNull(Whitebox.getInternalState(supplier, "activeClient")); + supplier.run(); + assertTrue(attempts.isEmpty()); + while (peer.getInputStream().read() != -1) { + // Drain the already-written request before observing socket closure. + } + } finally { + if (peer != null) { + peer.close(); + } + tasks.shutdownNow(); + assertTrue(tasks.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + public void testCloseAbortsBlockedDiscoveryQuery() throws Exception { + advertised = endpoints(0); + supplier.close(); + supplier = createSupplier(endpoints(0, 1, 2), 0); + supplier.run(); + TestNode node = nodes.get(0); + node.queryEntered = new CountDownLatch(1); + node.queryRelease = new CountDownLatch(1); + ExecutorService tasks = Executors.newCachedThreadPool(); + attempts.clear(); + try { + Future refresh = tasks.submit(supplier::run); + assertTrue(node.queryEntered.await(5, TimeUnit.SECONDS)); + tasks.submit(supplier::close).get(5, TimeUnit.SECONDS); + refresh.get(5, TimeUnit.SECONDS); + assertTrue(attempts.isEmpty()); + assertNull(Whitebox.getInternalState(supplier, "client")); + assertNull(Whitebox.getInternalState(supplier, "activeClient")); + } finally { + node.queryRelease.countDown(); + tasks.shutdownNow(); + assertTrue(tasks.awaitTermination(5, TimeUnit.SECONDS)); + } } @Test @@ -243,6 +439,7 @@ public void testQuerySelectionDuringEmptyDiscoveryRefresh() throws Exception { supplier .getQueryEndPoint() .ifPresent(endpoint -> assertEquals(nodes.get(0).endpoint, endpoint)); + assertTrue(supplier.getQueryEndPoint().isPresent()); } }); try { @@ -274,6 +471,10 @@ public void testRefreshRequestsAreCoalescedAndIgnoredAfterClose() { } private NodesSupplier createSupplier(List seeds) { + return createSupplier(seeds, 1000); + } + + private NodesSupplier createSupplier(List seeds, int timeout) { return NodesSupplier.createNodeSupplier( seeds, executor, @@ -282,7 +483,7 @@ private NodesSupplier createSupplier(List seeds) { ZoneId.of("UTC"), SessionConfig.DEFAULT_INITIAL_BUFFER_CAPACITY, SessionConfig.DEFAULT_MAX_FRAME_SIZE, - 1000, + timeout, false, null, null, @@ -332,6 +533,11 @@ private class TestNode implements AutoCloseable { private final ExecutorService workers = Executors.newCachedThreadPool(); private final Thread thread; private volatile boolean online = true; + private volatile boolean failQuery; + private volatile boolean queryApplicationError; + private volatile CountDownLatch queryEntered; + private volatile CountDownLatch queryRelease; + private final ThreadLocal activeTransport = new ThreadLocal<>(); private TestNode() throws Exception { ServerSocket listener = new ServerSocket(0); @@ -353,6 +559,20 @@ private TestNode() throws Exception { .thenAnswer( invocation -> { checkOnline(); + queries.add(endpoint); + if (queryEntered != null) { + queryEntered.countDown(); + queryRelease.await(); + } + if (failQuery) { + if (queryApplicationError) { + throw new TApplicationException( + TApplicationException.INTERNAL_ERROR, "Discovery query failed"); + } + return new TSExecuteStatementResp( + new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()) + .setMessage("Discovery query failed")); + } TsBlockBuilder block = new TsBlockBuilder(Arrays.asList(TSDataType.TEXT, TSDataType.INT32)); for (TEndPoint address : advertised) { @@ -380,6 +600,26 @@ private TestNode() throws Exception { .protocolFactory(new TBinaryProtocol.Factory()) .transportFactory(DeepCopyRpcTransportFactory.INSTANCE) .executorService(workers)); + server.setServerEventHandler( + new TServerEventHandler() { + @Override + public void preServe() {} + + @Override + public ServerContext createContext(TProtocol input, TProtocol output) { + return mock(ServerContext.class); + } + + @Override + public void processContext(ServerContext context, TTransport input, TTransport output) { + activeTransport.set(input); + } + + @Override + public void deleteContext(ServerContext context, TProtocol input, TProtocol output) { + activeTransport.remove(); + } + }); CountDownLatch ready = new CountDownLatch(1); thread = new Thread( @@ -395,6 +635,12 @@ private TestNode() throws Exception { private void checkOnline() throws TException { if (!online) { + // A stopped node breaks the transport; a reachable peer returning an RPC error is + // different. + TTransport current = activeTransport.get(); + if (current != null) { + current.close(); + } throw new TException("Node is restarting"); } } diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/ThriftConnectionLifecycleTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/ThriftConnectionLifecycleTest.java new file mode 100644 index 0000000000000..a702f0d3eee36 --- /dev/null +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/ThriftConnectionLifecycleTest.java @@ -0,0 +1,207 @@ +/* + * 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.iotdb.session; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.isession.SessionConfig; +import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.rpc.TElasticFramedTransport; + +import org.apache.thrift.transport.TSocket; +import org.apache.thrift.transport.TTransport; +import org.apache.thrift.transport.TTransportException; +import org.junit.Test; +import org.powermock.reflect.Whitebox; + +import java.net.Socket; +import java.net.SocketAddress; +import java.net.SocketException; +import java.time.ZoneId; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +public class ThriftConnectionLifecycleTest { + @Test + public void testAbortWhileTcpConnectIsBlocked() throws Exception { + assertTcpConnectAborted(false); + } + + @Test + public void testAbortBeforeTcpConnectCannotReopenSocket() throws Exception { + assertTcpConnectAborted(true); + } + + private void assertTcpConnectAborted(boolean beforeConnect) throws Exception { + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch proceed = new CountDownLatch(beforeConnect ? 1 : 0); + CountDownLatch closed = new CountDownLatch(1); + Socket socket = mock(Socket.class); + doAnswer( + invocation -> { + closed.countDown(); + return null; + }) + .when(socket) + .close(); + doAnswer( + invocation -> { + entered.countDown(); + assertTrue(closed.await(5, TimeUnit.SECONDS)); + throw new SocketException("Connection cancelled by test"); + }) + .when(socket) + .connect(any(SocketAddress.class), anyInt()); + TSocket raw = new TSocket("127.0.0.1", 1, 0); + raw.close(); + Whitebox.setInternalState(raw, "socket_", socket); + TTransport prepared = + new TElasticFramedTransport( + raw, + SessionConfig.DEFAULT_INITIAL_BUFFER_CAPACITY, + SessionConfig.DEFAULT_MAX_FRAME_SIZE, + true) { + @Override + public void open() throws TTransportException { + if (beforeConnect) { + entered.countDown(); + try { + assertTrue(proceed.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new TTransportException(e); + } + } + super.open(); + } + }; + ThriftConnection connection = connectionReturning(prepared, null, null); + ExecutorService tasks = Executors.newCachedThreadPool(); + try { + Future initialization = + tasks.submit( + () -> { + initialize(connection); + return null; + }); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + tasks.submit(connection::abort).get(5, TimeUnit.SECONDS); + assertTrue(closed.await(5, TimeUnit.SECONDS)); + if (beforeConnect) { + // TSocket.close() would null this field and allow open() to create another socket. + assertSame(socket, raw.getSocket()); + } + proceed.countDown(); + assertConnectionFailure(initialization); + verify(socket).connect(any(SocketAddress.class), anyInt()); + } finally { + closed.countDown(); + proceed.countDown(); + tasks.shutdownNow(); + assertTrue(tasks.awaitTermination(5, TimeUnit.SECONDS)); + connection.close(); + } + } + + @Test + public void testTransportCreatedAfterClosureIsDisposedWithoutOpening() throws Exception { + CountDownLatch creating = new CountDownLatch(1); + CountDownLatch publish = new CountDownLatch(1); + TTransport prepared = mock(TTransport.class); + ThriftConnection connection = connectionReturning(prepared, creating, publish); + ExecutorService tasks = Executors.newCachedThreadPool(); + try { + Future initialization = + tasks.submit( + () -> { + initialize(connection); + return null; + }); + assertTrue(creating.await(5, TimeUnit.SECONDS)); + tasks.submit(connection::close).get(5, TimeUnit.SECONDS); + publish.countDown(); + assertConnectionFailure(initialization); + verify(prepared).close(); + verify(prepared, never()).open(); + } finally { + publish.countDown(); + tasks.shutdownNow(); + assertTrue(tasks.awaitTermination(5, TimeUnit.SECONDS)); + connection.close(); + } + } + + private ThriftConnection connectionReturning( + TTransport prepared, CountDownLatch creating, CountDownLatch publish) { + return new ThriftConnection( + new TEndPoint("127.0.0.1", 1), + SessionConfig.DEFAULT_INITIAL_BUFFER_CAPACITY, + SessionConfig.DEFAULT_MAX_FRAME_SIZE, + 0) { + @Override + protected TTransport createTransport( + boolean ssl, + String trustStore, + String trustStorePwd, + String keyStore, + String keyStorePwd, + String protocol) + throws TTransportException { + if (creating != null) { + creating.countDown(); + try { + assertTrue(publish.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new TTransportException(e); + } + } + return prepared; + } + }; + } + + private void initialize(ThriftConnection connection) throws IoTDBConnectionException { + connection.init( + false, null, null, null, null, "TLS", "root", "root", false, ZoneId.of("UTC"), "V_1_0"); + } + + private void assertConnectionFailure(Future initialization) throws Exception { + try { + initialization.get(5, TimeUnit.SECONDS); + fail("A cancelled initialization must not succeed"); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof IoTDBConnectionException); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTask.java index 9c1311a77fe0a..4e12fdfffb969 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTask.java @@ -60,7 +60,9 @@ public static void buildTsBlock( String status = dataNodeInfo.getStatus(); if (NodeStatus.Removing.getStatus().equals(status) || NodeStatus.Unknown.getStatus().equals(status) - || NodeStatus.Stopped.getStatus().equals(status)) { + || NodeStatus.Stopped.getStatus().equals(status) + || (NodeStatus.ReadOnly.getStatus().equals(status) + && NodeStatus.STOPPING.equals(dataNodeInfo.getStatusReason()))) { continue; } builder.getTimeColumnBuilder().writeLong(0L); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTaskTest.java index 734915bcff9d4..94344f4bed89f 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTaskTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTaskTest.java @@ -69,6 +69,29 @@ public void testNoOnlineNodes() throws Exception { assertEquals(0, execute(null).getPositionCount()); } + @Test + public void testReadOnlyStoppingIsExcludedButOtherReasonsRemainReadable() throws Exception { + List nodes = + Arrays.asList( + readOnlyNode(6667, NodeStatus.STOPPING), + readOnlyNode(6668, NodeStatus.DISK_FULL), + readOnlyNode(6669, NodeStatus.MANUAL), + readOnlyNode(6670, null)); + TsBlock result = execute(nodes); + assertEquals(3, result.getPositionCount()); + for (int i = 0; i < 3; i++) { + assertEquals(6668 + i, result.getColumn(1).getInt(i)); + } + } + + private TDataNodeInfo readOnlyNode(int port, String reason) { + return new TDataNodeInfo() + .setStatus(NodeStatus.ReadOnly.getStatus()) + .setStatusReason(reason) + .setRpcAddresss("127.0.0.1") + .setRpcPort(port); + } + private TsBlock execute(List nodes) throws Exception { SettableFuture future = SettableFuture.create(); ShowAvailableUrlsTask.buildTsBlock( From ab19eeb3b64a46d298bc0bd9642d219e333984d9 Mon Sep 17 00:00:00 2001 From: JackieTien97 Date: Tue, 20 Oct 2026 14:08:40 +0800 Subject: [PATCH 3/3] Avoid pinning node discovery after query errors Close query-failed connections and defer their endpoints across refresh rounds while preserving the last usable snapshot and bounded query retries. Retain all initial addresses when reopening a Session without discovery. Poll for integration-test snapshots and keep upgrade guidance in the PR. --- RELEASE_NOTES.md | 26 ---- .../session/it/SessionSeedNodeFallbackIT.java | 40 ++++-- .../apache/iotdb/session/NodesSupplier.java | 26 +++- .../org/apache/iotdb/session/Session.java | 6 +- .../session/SessionSeedNodeFallbackTest.java | 128 +++++++++++++++--- 5 files changed, 162 insertions(+), 64 deletions(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 851a79f429b9f..96b941b107d18 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -19,32 +19,6 @@ --> -# Unreleased - -## Java client discovery and rolling upgrades - -`SHOW AVAILABLE URLS` excludes `Unknown`, `Stopped`, and `Removing` DataNodes, as well as -`ReadOnly` nodes whose status reason is `Stopping`. Other `ReadOnly` nodes remain query endpoints. -The Java SDK keeps its initial `nodeUrls` as recovery seeds when the latest discovered endpoints -cannot be reached. Configure several stable seeds and keep at least one reachable during a rolling -upgrade. - -Upgrade Java clients to a version with seed fallback **before upgrading the servers**. The server -filter applies to every client. Older Java SDKs and the C++ client do not gain seed fallback from a -server upgrade: their discovered list may lose a node that later becomes the only reachable node. -For those clients, keep an endpoint in their last discovery snapshot reachable throughout maintenance, -or use a stable explicit endpoint list with automatic discovery disabled until the client supports -seed fallback. This change does not port the Java recovery mechanism to C++. - -Seed and discovered endpoints are deduplicated by their literal host/IP and port, without DNS -resolution. Configure `nodeUrls` using the same address form as each DataNode's advertised -`dn_rpc_address`; for example, do not mix `localhost` seeds with `127.0.0.1` discovery addresses for -the same node. For wildcard bind addresses such as `0.0.0.0`, supply reachable addresses in -`nodeUrls`; wildcard discovery rows do not clear the previous snapshot. Aliases can otherwise cause -duplicate connection attempts, extra discovery refreshes, -and reconnections on retryable responses. DNS names may represent multiple or changing addresses, -so recovery does not resolve them to infer permanent node identity. - # Apache IoTDB 2.0.10 ## Features & Improvements 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 81df4e4fcf114..a13fc8183b919 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 @@ -35,6 +35,7 @@ import org.apache.iotdb.session.Session; import org.apache.iotdb.session.pool.SessionPool; +import org.awaitility.Awaitility; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -94,21 +95,21 @@ public void testQueriesDuringRollingRestart() throws Exception { assertQuery(session, pool); NodesSupplier sessionSupplier = supplier(session); NodesSupplier poolSupplier = supplier(pool); + Set runningNodes = + new HashSet<>(Arrays.asList(endpoint(0), endpoint(1), endpoint(2))); for (int index = 0; index < 3; index++) { EnvFactory.getEnv().shutdownDataNode(index); awaitStatus(index, EnvUtils.getNodeStatusAfterLocalStop()); - sessionSupplier.run(); - poolSupplier.run(); - assertFalse(sessionSupplier.get().contains(endpoint(index))); - assertFalse(poolSupplier.get().contains(endpoint(index))); + runningNodes.remove(endpoint(index)); + awaitSnapshot(sessionSupplier, runningNodes); + awaitSnapshot(poolSupplier, runningNodes); assertQuery(session, pool); EnvFactory.getEnv().startDataNode(index); awaitStatus(index, NodeStatus.Running); - sessionSupplier.run(); - poolSupplier.run(); - assertTrue(sessionSupplier.get().contains(endpoint(index))); - assertTrue(poolSupplier.get().contains(endpoint(index))); + runningNodes.add(endpoint(index)); + awaitSnapshot(sessionSupplier, runningNodes); + awaitSnapshot(poolSupplier, runningNodes); assertQuery(session, pool); } } finally { @@ -139,11 +140,9 @@ public ScheduledFuture scheduleAtFixedRate( NodesSupplier business = createSupplier(executor); NodesSupplier background = createSupplier(executor); try (Session session = new Session.Builder().nodeUrls(urls()).enableAutoFetch(false).build()) { - business.run(); - background.run(); - assertEquals( - new HashSet<>(Arrays.asList(endpoint(1), endpoint(2))), new HashSet<>(business.get())); - assertEquals(business.get(), background.get()); + Set discovered = new HashSet<>(Arrays.asList(endpoint(1), endpoint(2))); + awaitSnapshot(business, discovered); + awaitSnapshot(background, discovered); session.open(false, 1000, new ConcurrentHashMap<>(), business); EnvFactory.getEnv().startDataNode(0); @@ -159,8 +158,7 @@ public ScheduledFuture scheduleAtFixedRate( try (SessionDataSet result = session.executeQueryStatement("SHOW AVAILABLE URLS")) { assertEquals(Collections.singleton(endpoint(0)), readEndpoints(result)); } - background.run(); - assertEquals(Collections.singletonList(endpoint(0)), background.get()); + awaitSnapshot(background, Collections.singleton(endpoint(0))); assertEquals(3, background.getSeedNodes().size()); // Wait for the immediate refresh requested by business reconnection, not a 60-second poll. executor.submit(() -> {}).get(10, TimeUnit.SECONDS); @@ -241,4 +239,16 @@ private void awaitStatus(int index, NodeStatus status) { Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(index)), Collections.singletonList(status)); } + + private void awaitSnapshot(NodesSupplier current, Set expected) { + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .pollInterval(100, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> { + // run() can return immediately while a scheduled refresh is still in flight. + current.run(); + assertEquals(expected, new HashSet<>(current.get())); + }); + } } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java index 572050edeaf3e..fe91277a27dc8 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java @@ -96,6 +96,8 @@ public class NodesSupplier implements INodeSupplier, Runnable { // Refresh serialization is independent of the short client-publication/cleanup critical section. private final ReentrantLock refreshLock = new ReentrantLock(); + // Accessed under refreshLock, in failure order, until a discovery query succeeds. + private final Set queryFailedNodes = new LinkedHashSet<>(); private final Object clientLock = new Object(); private ThriftConnection client; private ThriftConnection activeClient; @@ -288,6 +290,11 @@ public void run() { Collections.rotate(seeds, -ThreadLocalRandom.current().nextInt(seeds.size())); } candidates.addAll(seeds); + // Defer query-failed nodes across refresh rounds without losing them as recovery candidates. + // Remember all failures so two failing nodes cannot starve a third, healthy candidate. + queryFailedNodes.retainAll(candidates); + candidates.removeAll(queryFailedNodes); + candidates.addAll(queryFailedNodes); for (TEndPoint endPoint : candidates) { if (closed) { return; @@ -319,13 +326,21 @@ public void run() { } } - /** Return false only when the connection failed and another candidate should be tried. */ + /** Return false only when the transport failed and another candidate should be tried now. */ private boolean refreshFrom(ThriftConnection current) { if (closed) { return true; } try { - updateDataNodeList(current); + if (updateDataNodeList(current)) { + queryFailedNodes.clear(); + } else { + // End this round on a query error, but try another candidate on the next refresh. + // Moving repeated failures to the end also permits recovery after a cluster-wide outage. + queryFailedNodes.remove(current.endPoint); + queryFailedNodes.add(current.endPoint); + destroyClient(current); + } } catch (IoTDBConnectionException e) { destroyClient(current); return false; @@ -409,7 +424,7 @@ public Optional getQueryEndPoint() { } } - private void updateDataNodeList(ThriftConnection current) throws IoTDBConnectionException { + private boolean updateDataNodeList(ThriftConnection current) throws IoTDBConnectionException { try (SessionDataSet sessionDataSet = current.executeQueryStatement(SHOW_AVAILABLE_URLS_COMMAND, TIMEOUT_IN_MS, FETCH_SIZE)) { updateAvailableNodes(sessionDataSet); @@ -421,13 +436,14 @@ private void updateDataNodeList(ThriftConnection current) throws IoTDBConnection throw e; } // A Thrift application error also means the peer responded; it is not a transport failure. + return false; } catch (Exception e) { - // A query error (for example no ConfigNode leader) is not evidence of a dead endpoint. - // Keep this connection and snapshot, and let the next scheduled refresh retry the query. if (!closed) { LOGGER.warn(SessionMessages.FAILED_TO_FETCH_DATA_NODE_LIST, current.endPoint); } + return false; } + return true; } private void updateAvailableNodes(SessionDataSet sessionDataSet) throws Exception { diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java index 9ac9c9d8cb9d5..e5d44af969033 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java @@ -88,6 +88,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Map.Entry; @@ -563,7 +564,10 @@ public synchronized void open(boolean enableRPCCompaction, int connectionTimeout enableRPCCompaction, version.toString()); } else { - this.availableNodes = new DummyNodesSupplier(getNodeUrls()); + // A static supplier has no separate seed tier. Keep every configured address after reopen. + Set staticNodes = new LinkedHashSet<>(getNodeUrls()); + staticNodes.addAll(SessionUtils.parseSeedNodeUrls(initialNodeUrls)); + this.availableNodes = new DummyNodesSupplier(new ArrayList<>(staticNodes)); } this.connectionTimeoutInMs = connectionTimeoutInMs; diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java index b30d6a9a3367d..24f7e6ab7cb27 100644 --- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java @@ -21,6 +21,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.isession.pool.SessionDataSetWrapper; @@ -204,44 +205,105 @@ public void testWildcardDiscoveryKeepsSeedsAndLastSnapshot() { } @Test - public void testQueryStatusFailureDoesNotRotateConnections() { - assertQueryFailureDoesNotRotate(false); + public void testQueryStatusFailureMovesDiscoveryToAnotherNodeNextRound() { + assertQueryFailureMovesDiscovery(false); } @Test - public void testQueryApplicationFailureDoesNotRotateConnections() { - assertQueryFailureDoesNotRotate(true); + public void testQueryApplicationFailureMovesDiscoveryToAnotherNodeNextRound() { + assertQueryFailureMovesDiscovery(true); } - private void assertQueryFailureDoesNotRotate(boolean applicationError) { - advertised = endpoints(1, 2); + private void assertQueryFailureMovesDiscovery(boolean applicationError) { supplier.run(); - nodes.forEach( - node -> { - node.failQuery = true; - node.queryApplicationError = applicationError; - }); + nodes.get(0).failQuery = true; + nodes.get(0).queryApplicationError = applicationError; + advertised = endpoints(1, 2); queries.clear(); attempts.clear(); supplier.run(); assertEquals(endpoints(0), queries); assertTrue(attempts.isEmpty()); + assertEquals(endpoints(0, 1, 2), supplier.get()); + + // Node 0 is still reachable but cannot query the ConfigNode. Do not pin discovery to it. + supplier.run(); + assertEquals(endpoints(0, 1), queries); + assertEquals(endpoints(1), attempts); assertEquals(endpoints(1, 2), supplier.get()); - nodes.forEach(node -> node.failQuery = false); advertised = endpoints(2); supplier.run(); assertEquals(endpoints(2), supplier.get()); - assertTrue(attempts.isEmpty()); + assertEquals(endpoints(0, 1, 1), queries); + assertEquals(endpoints(1), attempts); + } + + @Test + public void testClusterQueryStatusFailuresAreBounded() { + assertClusterQueryFailuresAreBounded(false); + } + + @Test + public void testClusterQueryApplicationFailuresAreBounded() { + assertClusterQueryFailuresAreBounded(true); + } + + private void assertClusterQueryFailuresAreBounded(boolean applicationError) { + nodes.forEach( + node -> { + node.failQuery = true; + node.queryApplicationError = applicationError; + }); + for (int i = 0; i < 6; i++) { + supplier.run(); + // One new session and one query per refresh, even when every node returns an error. + assertEquals(i + 1, attempts.size()); + assertEquals(i + 1, queries.size()); + assertEquals(nodes.get(i % 3).endpoint, queries.get(i)); + assertEquals(endpoints(0, 1, 2), supplier.get()); + } + assertEquals(queries, attempts); + + // The first two candidates still fail. Discovery must eventually reach the recovered third. + nodes.get(2).failQuery = false; + advertised = endpoints(2); + for (int i = 0; i < 3; i++) { + supplier.run(); + assertEquals(7 + i, attempts.size()); + assertEquals(7 + i, queries.size()); + } + assertEquals(endpoints(2), supplier.get()); + supplier.run(); + assertEquals(9, attempts.size()); + assertEquals(10, queries.size()); + assertEquals(nodes.get(2).endpoint, queries.get(9)); } @Test - public void testQueryFailureAfterConnectingDoesNotTryOtherCandidates() { - nodes.forEach(node -> node.failQuery = true); + public void testDiscoveryQueryFailuresEventuallyReachRecoverySeed() { + advertised = endpoints(0, 1); + supplier.run(); + nodes.get(0).failQuery = true; + nodes.get(1).failQuery = true; + advertised = endpoints(2); + queries.clear(); + attempts.clear(); supplier.run(); - assertEquals(endpoints(0), attempts); assertEquals(endpoints(0), queries); - assertEquals(endpoints(0, 1, 2), supplier.get()); + assertTrue(attempts.isEmpty()); + assertEquals(endpoints(0, 1), supplier.get()); + + supplier.run(); + assertEquals(endpoints(0, 1), queries); + assertEquals(endpoints(1), attempts); + assertEquals(endpoints(0, 1), supplier.get()); + + supplier.run(); + assertEquals(endpoints(0, 1, 2), queries); + assertEquals(endpoints(1, 2), attempts); + assertEquals(endpoints(2), supplier.get()); + assertEquals(endpoints(0, 1, 2), supplier.getSeedNodes()); } @Test @@ -296,6 +358,38 @@ public void testSingleHostSessionReopenAfterFailover() throws Exception { assertSessionReopenAfterFailover(true); } + @Test + public void testStaticNodeUrlsSurviveReopenAndSecondFailover() throws Exception { + nodes.get(1).online = false; + try (Session session = + new Session.Builder() + .nodeUrls(Arrays.asList(url(0), url(1), url(0))) + .enableAutoFetch(false) + .build()) { + session.open(false, 1000); + assertEquals(nodes.get(0).endpoint, session.defaultEndPoint); + nodes.get(0).online = false; + nodes.get(1).online = true; + assertTrue(reconnect(session.defaultSessionConnection)); + assertEquals(nodes.get(1).endpoint, session.defaultEndPoint); + session.close(); + + session.open(false, 1000); + INodeSupplier reopened = Whitebox.getInternalState(session, "availableNodes"); + assertEquals(endpoints(1, 0), reopened.get()); + nodes.get(1).online = false; + nodes.get(0).online = true; + assertTrue(reconnect(session.defaultSessionConnection)); + assertEquals(nodes.get(0).endpoint, session.defaultEndPoint); + advertised = endpoints(0); + try (SessionDataSet result = session.executeQueryStatement("SHOW AVAILABLE URLS")) { + assertTrue(result.hasNext()); + result.next(); + assertFalse(result.hasNext()); + } + } + } + private void assertSessionReopenAfterFailover(boolean singleHost) throws Exception { List urls = new ArrayList<>(Collections.singletonList(url(0))); Session session =