This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 85277b3b689 Filter offline DataNode URLs and add Java client seed
fallback (#18921)
85277b3b689 is described below
commit 85277b3b689f4a9cf7a1bd779a57c09c9deeb588
Author: Jackie Tien <[email protected]>
AuthorDate: Tue Oct 20 16:41:17 2026 +0800
Filter offline DataNode URLs and add Java client seed fallback (#18921)
---
.../session/it/SessionSeedNodeFallbackIT.java | 254 +++++++
.../org/apache/iotdb/isession/INodeSupplier.java | 11 +
.../org/apache/iotdb/session/NodesSupplier.java | 276 ++++++--
.../java/org/apache/iotdb/session/Session.java | 31 +-
.../apache/iotdb/session/SessionConnection.java | 31 +-
.../apache/iotdb/session/TableSessionBuilder.java | 3 +
.../org/apache/iotdb/session/ThriftConnection.java | 127 +++-
.../org/apache/iotdb/session/pool/SessionPool.java | 13 +-
.../session/pool/TableSessionPoolBuilder.java | 3 +
.../iotdb/session/SessionSeedNodeFallbackTest.java | 751 +++++++++++++++++++++
.../session/ThriftConnectionLifecycleTest.java | 207 ++++++
.../config/metadata/ShowAvailableUrlsTask.java | 8 +-
.../config/metadata/ShowAvailableUrlsTaskTest.java | 101 +++
13 files changed, 1715 insertions(+), 101 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/session/it/SessionSeedNodeFallbackIT.java
b/integration-test/src/test/java/org/apache/iotdb/session/it/SessionSeedNodeFallbackIT.java
new file mode 100644
index 00000000000..a13fc8183b9
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/session/it/SessionSeedNodeFallbackIT.java
@@ -0,0 +1,254 @@
+/*
+ * 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.awaitility.Awaitility;
+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);
+ Set<TEndPoint> 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());
+ runningNodes.remove(endpoint(index));
+ awaitSnapshot(sessionSupplier, runningNodes);
+ awaitSnapshot(poolSupplier, runningNodes);
+ assertQuery(session, pool);
+
+ EnvFactory.getEnv().startDataNode(index);
+ awaitStatus(index, NodeStatus.Running);
+ runningNodes.add(endpoint(index));
+ awaitSnapshot(sessionSupplier, runningNodes);
+ awaitSnapshot(poolSupplier, runningNodes);
+ 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()) {
+ Set<TEndPoint> 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);
+ 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));
+ }
+ 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);
+ 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<TEndPoint> 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<TEndPoint> readEndpoints(SessionDataSet dataSet) throws
Exception {
+ Set<TEndPoint> 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<String> 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));
+ }
+
+ private void awaitSnapshot(NodesSupplier current, Set<TEndPoint> 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/isession/src/main/java/org/apache/iotdb/isession/INodeSupplier.java
b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/INodeSupplier.java
index 60a2a5adb43..9c8d32f9461 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 @@ package org.apache.iotdb.isession;
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<List<TEndPoint>> {
void close();
Optional<TEndPoint> getQueryEndPoint();
+
+ /** Initial addresses to try after all discovered endpoints have failed. */
+ default List<TEndPoint> 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 81d83450f72..fe91277a27d 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,20 +22,29 @@ package org.apache.iotdb.session;
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;
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.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 {
@@ -54,10 +63,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<TEndPoint> availableNodes = new
CopyOnWriteArrayList<>();
+ private volatile List<TEndPoint> availableNodes;
+
+ private final List<TEndPoint> seedNodes;
+ private final ScheduledExecutorService executorService;
+ private final AtomicBoolean refreshRequested = new AtomicBoolean();
private final boolean useSSL;
private final String trustStore;
@@ -82,13 +94,59 @@ 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();
+ // Accessed under refreshLock, in failure order, until a discovery query
succeeds.
+ private final Set<TEndPoint> queryFailedNodes = new LinkedHashSet<>();
+ private final Object clientLock = new Object();
private ThriftConnection client;
+ private ThriftConnection activeClient;
private volatile boolean closed = false;
+ public static NodesSupplier createNodeSupplier(
+ List<TEndPoint> 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<TEndPoint> endPointList,
+ List<TEndPoint> seedNodes,
ScheduledExecutorService executorService,
String userName,
String password,
@@ -108,6 +166,8 @@ public class NodesSupplier implements INodeSupplier,
Runnable {
NodesSupplier nodesSupplier =
new NodesSupplier(
endPointList,
+ seedNodes,
+ executorService,
userName,
password,
zoneId,
@@ -133,6 +193,8 @@ public class NodesSupplier implements INodeSupplier,
Runnable {
private NodesSupplier(
List<TEndPoint> endPointList,
+ List<TEndPoint> seedNodes,
+ ScheduledExecutorService executorService,
String userName,
String password,
ZoneId zoneId,
@@ -147,7 +209,13 @@ public class NodesSupplier implements INodeSupplier,
Runnable {
String sslProtocol,
boolean enableRPCCompression,
String version) {
- this.availableNodes.addAll(new HashSet<>(endPointList));
+ 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.executorService = executorService;
this.userName = userName;
this.password = password;
this.useSSL = useSSL;
@@ -164,43 +232,135 @@ public class NodesSupplier implements INodeSupplier,
Runnable {
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<TEndPoint> get() {
return availableNodes;
}
@Override
- public void run() {
- if (closed) {
- if (client != null) {
- destroyCurrentClient();
+ public List<TEndPoint> 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 void run() {
+ if (closed || !refreshLock.tryLock()) {
return;
}
- if (client == null) {
- for (TEndPoint endPoint : availableNodes) {
- if (createConnection(endPoint)) {
- break;
+ try {
+ Set<TEndPoint> 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;
}
}
+
+ Set<TEndPoint> candidates = new LinkedHashSet<>(availableNodes);
+ List<TEndPoint> seeds = new ArrayList<>(seedNodes);
+ if (!seeds.isEmpty()) {
+ 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;
+ }
+ 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();
+ }
}
+ }
- // make sure the following code block can run thread-safely with close()
method.
- synchronized (this) {
- if (client != null && !updateDataNodeList()) {
- destroyCurrentClient();
+ /** Return false only when the transport failed and another candidate should
be tried now. */
+ private boolean refreshFrom(ThriftConnection current) {
+ if (closed) {
+ return true;
+ }
+ try {
+ 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;
}
+ 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,
@@ -212,42 +372,75 @@ public class NodesSupplier implements INodeSupplier,
Runnable {
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
public Optional<TEndPoint> getQueryEndPoint() {
- if (availableNodes == null || availableNodes.isEmpty()) {
+ List<TEndPoint> snapshot = availableNodes;
+ if (snapshot.isEmpty()) {
return Optional.empty();
} else {
- return Optional.of(policy.chooseOne(get()));
+ return Optional.of(policy.chooseOne(snapshot));
}
}
- private boolean updateDataNodeList() {
+ private boolean 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);
+ } 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.
+ return false;
+ } catch (Exception e) {
+ if (!closed) {
+ LOGGER.warn(SessionMessages.FAILED_TO_FETCH_DATA_NODE_LIST,
current.endPoint);
+ }
return false;
}
return true;
@@ -266,9 +459,10 @@ public class NodesSupplier implements INodeSupplier,
Runnable {
}
}
}
- // replace the older ones.
- if (!res.isEmpty()) {
- availableNodes = 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 a6deda17491..e5d44af9690 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.Collections;
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;
@@ -125,6 +126,7 @@ public class Session implements ISession {
new
LinkedBlockingQueue<>(SessionConfig.DEFAULT_SESSION_EXECUTOR_TASK_NUM),
ThreadUtils.createThreadFactory("SessionExecutor", true));
protected List<String> nodeUrls;
+ private final List<String> initialNodeUrls;
protected String username;
protected String password;
protected boolean useEncryptedPassword;
@@ -367,6 +369,8 @@ public class Session implements ISession {
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 +438,9 @@ public class Session implements ISession {
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 +456,13 @@ public class Session implements ISession {
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;
@@ -540,6 +547,7 @@ public class Session implements ISession {
this.availableNodes =
NodesSupplier.createNodeSupplier(
getNodeUrls(),
+ SessionUtils.parseSeedNodeUrls(initialNodeUrls),
executorService,
username,
password,
@@ -556,7 +564,10 @@ public class Session implements ISession {
enableRPCCompaction,
version.toString());
} else {
- this.availableNodes = new DummyNodesSupplier(getNodeUrls());
+ // A static supplier has no separate seed tier. Keep every configured
address after reopen.
+ Set<TEndPoint> staticNodes = new LinkedHashSet<>(getNodeUrls());
+ staticNodes.addAll(SessionUtils.parseSeedNodeUrls(initialNodeUrls));
+ this.availableNodes = new DummyNodesSupplier(new
ArrayList<>(staticNodes));
}
this.connectionTimeoutInMs = connectionTimeoutInMs;
@@ -599,11 +610,9 @@ public class Session implements ISession {
}
private List<TEndPoint> getNodeUrls() {
- if (defaultEndPoint != null) {
- return Collections.singletonList(defaultEndPoint);
- } else {
- return SessionUtils.parseSeedNodeUrls(nodeUrls);
- }
+ return defaultEndPoint == null
+ ? SessionUtils.parseSeedNodeUrls(initialNodeUrls)
+ : Collections.singletonList(defaultEndPoint.deepCopy());
}
@Override
@@ -4491,6 +4500,10 @@ public class Session implements ISession {
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<String> 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 7b6ec3b03df..ce06b35abb1 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 @@ package org.apache.iotdb.session;
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.security.SecureRandom;
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,22 @@ public class SessionConnection {
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;
+ List<TEndPoint> discoveredNodes = availableNodes.get();
+ endPointList = new ArrayList<>(discoveredNodes);
+ if (!endPointList.isEmpty()) {
+ Collections.rotate(endPointList,
-random.nextInt(endPointList.size()));
+ }
+ if (availableNodes instanceof INodeSupplier) {
+ List<TEndPoint> seeds = new ArrayList<>(((INodeSupplier)
availableNodes).getSeedNodes());
+ if (!seeds.isEmpty()) {
+ Collections.rotate(seeds, -random.nextInt(seeds.size()));
}
- tryHostNum++;
+ endPointList.addAll(seeds);
+ }
+ // 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 +1171,9 @@ public class SessionConnection {
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/TableSessionBuilder.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java
index 20147e1cfcf..a39af4c847a 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.
*
+ * <p>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 4f8e2a58af7..71cc63526d7 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.DeepCopyRpcTransportFactory;
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.iotdb.session.i18n.SessionMessages;
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 class ThriftConnection {
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 class ThriftConnection {
statementId = client.requestStatementId(sessionId);
} catch (Exception e) {
- transport.close();
+ close();
throw new IoTDBConnectionException(e);
}
}
@@ -193,19 +197,74 @@ public class ThriftConnection {
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 32feac2cea5..16b79c6415c 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 org.slf4j.LoggerFactory;
import java.io.IOException;
import java.time.ZoneId;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -502,7 +503,7 @@ public class SessionPool implements ISessionPool {
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;
@@ -519,9 +520,9 @@ public class SessionPool implements ISessionPool {
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) {
@@ -564,7 +565,7 @@ public class SessionPool implements ISessionPool {
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();
@@ -3697,6 +3698,10 @@ public class SessionPool implements ISessionPool {
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<String> 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 790e7816288..f19c5fc8878 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.
*
+ * <p>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
new file mode 100644
index 00000000000..24f7e6ab7cb
--- /dev/null
+++
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionSeedNodeFallbackTest.java
@@ -0,0 +1,751 @@
+/*
+ * 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.INodeSupplier;
+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;
+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.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;
+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.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;
+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.assertNull;
+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<TestNode> nodes = new ArrayList<>();
+ private final List<TEndPoint> attempts = new CopyOnWriteArrayList<>();
+ private final List<TEndPoint> queries = new CopyOnWriteArrayList<>();
+ private final List<Runnable> refreshes = new ArrayList<>();
+ private final List<SessionConnection> connections = new ArrayList<>();
+ private volatile List<TEndPoint> 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 testEmptyDiscoveryKeepsPreviousSnapshot() {
+ advertised = endpoints(1, 2);
+ supplier.run();
+ attempts.clear();
+ advertised = Collections.emptyList();
+ supplier.run();
+ 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());
+ 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<TEndPoint> 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 testQueryStatusFailureMovesDiscoveryToAnotherNodeNextRound() {
+ assertQueryFailureMovesDiscovery(false);
+ }
+
+ @Test
+ public void
testQueryApplicationFailureMovesDiscoveryToAnotherNodeNextRound() {
+ assertQueryFailureMovesDiscovery(true);
+ }
+
+ private void assertQueryFailureMovesDiscovery(boolean applicationError) {
+ supplier.run();
+ 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());
+
+ advertised = endpoints(2);
+ supplier.run();
+ assertEquals(endpoints(2), supplier.get());
+ 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 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), queries);
+ 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
+ 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++) {
+ List<TEndPoint> 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
+ public void testSeedSnapshotIsIndependentOfCallerAndDiscovery() {
+ supplier.close();
+ List<TEndPoint> 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 testNodeUrlsSessionReopenAfterFailover() throws Exception {
+ assertSessionReopenAfterFailover(false);
+ }
+
+ @Test
+ 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<String> 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();
+ 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<TEndPoint> 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<String> 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<TEndPoint> seeds = new ArrayList<>();
+ seeds.add(new TEndPoint("127.0.0.1", listener.getLocalPort()));
+ seeds.addAll(endpoints(1, 2));
+ supplier = createSupplier(seeds, 0);
+ Future<Socket> 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
+ 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));
+ assertTrue(supplier.getQueryEndPoint().isPresent());
+ }
+ });
+ 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<TEndPoint> seeds) {
+ return createSupplier(seeds, 1000);
+ }
+
+ private NodesSupplier createSupplier(List<TEndPoint> seeds, int timeout) {
+ return NodesSupplier.createNodeSupplier(
+ seeds,
+ executor,
+ "root",
+ "root",
+ ZoneId.of("UTC"),
+ SessionConfig.DEFAULT_INITIAL_BUFFER_CAPACITY,
+ SessionConfig.DEFAULT_MAX_FRAME_SIZE,
+ timeout,
+ 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<TEndPoint> endpoints(int... indices) {
+ List<TEndPoint> 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 volatile boolean failQuery;
+ private volatile boolean queryApplicationError;
+ private volatile CountDownLatch queryEntered;
+ private volatile CountDownLatch queryRelease;
+ private final ThreadLocal<TTransport> activeTransport = new
ThreadLocal<>();
+
+ 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();
+ 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) {
+ 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));
+ 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(
+ () -> {
+ 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) {
+ // 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");
+ }
+ }
+
+ @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-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 00000000000..a702f0d3eee
--- /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 6d957769b44..4e12fdfffb9 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,11 @@ public class ShowAvailableUrlsTask implements IConfigTask {
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)
+ || (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
new file mode 100644
index 00000000000..94344f4bed8
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowAvailableUrlsTaskTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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<TDataNodeInfo> 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());
+ }
+
+ @Test
+ public void testReadOnlyStoppingIsExcludedButOtherReasonsRemainReadable()
throws Exception {
+ List<TDataNodeInfo> 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<TDataNodeInfo> nodes) throws Exception {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ ShowAvailableUrlsTask.buildTsBlock(
+ new TShowDataNodesResp().setDataNodesInfoList(nodes), future);
+ return future.get().getResultSet();
+ }
+}
