Skip to content

Commit 85277b3

Browse files
JackieTien97JackieTien97
authored
Filter offline DataNode URLs and add Java client seed fallback (#18921)
1 parent 49ba3be commit 85277b3

13 files changed

Lines changed: 1715 additions & 101 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/session/it/SessionSeedNodeFallbackIT.java‎

Lines changed: 254 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,254 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.session.it;
21+
22+
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
23+
import org.apache.iotdb.commons.cluster.NodeStatus;
24+
import org.apache.iotdb.consensus.ConsensusFactory;
25+
import org.apache.iotdb.isession.SessionConfig;
26+
import org.apache.iotdb.isession.SessionDataSet;
27+
import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
28+
import org.apache.iotdb.it.env.EnvFactory;
29+
import org.apache.iotdb.it.env.cluster.EnvUtils;
30+
import org.apache.iotdb.it.framework.IoTDBTestRunner;
31+
import org.apache.iotdb.itbase.category.ClusterIT;
32+
import org.apache.iotdb.rpc.IoTDBConnectionException;
33+
import org.apache.iotdb.rpc.UrlUtils;
34+
import org.apache.iotdb.session.NodesSupplier;
35+
import org.apache.iotdb.session.Session;
36+
import org.apache.iotdb.session.pool.SessionPool;
37+
38+
import org.awaitility.Awaitility;
39+
import org.junit.After;
40+
import org.junit.Before;
41+
import org.junit.Test;
42+
import org.junit.experimental.categories.Category;
43+
import org.junit.runner.RunWith;
44+
45+
import java.lang.reflect.Field;
46+
import java.time.ZoneId;
47+
import java.util.ArrayList;
48+
import java.util.Arrays;
49+
import java.util.Collections;
50+
import java.util.HashSet;
51+
import java.util.List;
52+
import java.util.Set;
53+
import java.util.concurrent.ConcurrentHashMap;
54+
import java.util.concurrent.ScheduledFuture;
55+
import java.util.concurrent.ScheduledThreadPoolExecutor;
56+
import java.util.concurrent.TimeUnit;
57+
import java.util.stream.Collectors;
58+
59+
import static org.junit.Assert.assertEquals;
60+
import static org.junit.Assert.assertFalse;
61+
import static org.junit.Assert.assertTrue;
62+
63+
@RunWith(IoTDBTestRunner.class)
64+
@Category(ClusterIT.class)
65+
public class SessionSeedNodeFallbackIT {
66+
@Before
67+
public void setUp() {
68+
EnvFactory.getEnv()
69+
.getConfig()
70+
.getCommonConfig()
71+
.setSchemaReplicationFactor(3)
72+
.setDataReplicationFactor(3)
73+
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
74+
.setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
75+
EnvFactory.getEnv().initClusterEnvironment(1, 3);
76+
}
77+
78+
@After
79+
public void tearDown() {
80+
EnvFactory.getEnv().cleanClusterEnvironment();
81+
}
82+
83+
@Test
84+
public void testQueriesDuringRollingRestart() throws Exception {
85+
SessionPool pool =
86+
new SessionPool.Builder().nodeUrls(urls()).maxSize(1).connectionTimeoutInMs(1000).build();
87+
Session session = new Session.Builder().nodeUrls(urls()).build();
88+
try {
89+
// Initial Ratis region creation can exceed one second; do not retry its DDL prematurely.
90+
session.open(false, 30_000);
91+
session.executeNonQueryStatement("CREATE DATABASE root.seed_fallback");
92+
session.executeNonQueryStatement(
93+
"CREATE TIMESERIES root.seed_fallback.d.s WITH DATATYPE=INT32, ENCODING=RLE");
94+
session.executeNonQueryStatement("INSERT INTO root.seed_fallback.d(time,s) VALUES(1,42)");
95+
assertQuery(session, pool);
96+
NodesSupplier sessionSupplier = supplier(session);
97+
NodesSupplier poolSupplier = supplier(pool);
98+
Set<TEndPoint> runningNodes =
99+
new HashSet<>(Arrays.asList(endpoint(0), endpoint(1), endpoint(2)));
100+
for (int index = 0; index < 3; index++) {
101+
EnvFactory.getEnv().shutdownDataNode(index);
102+
awaitStatus(index, EnvUtils.getNodeStatusAfterLocalStop());
103+
runningNodes.remove(endpoint(index));
104+
awaitSnapshot(sessionSupplier, runningNodes);
105+
awaitSnapshot(poolSupplier, runningNodes);
106+
assertQuery(session, pool);
107+
108+
EnvFactory.getEnv().startDataNode(index);
109+
awaitStatus(index, NodeStatus.Running);
110+
runningNodes.add(endpoint(index));
111+
awaitSnapshot(sessionSupplier, runningNodes);
112+
awaitSnapshot(poolSupplier, runningNodes);
113+
assertQuery(session, pool);
114+
}
115+
} finally {
116+
try {
117+
session.close();
118+
} catch (IoTDBConnectionException e) {
119+
// Cached connections may still belong to a stopped process. Only cleanup tolerates this;
120+
// all query and discovery assertions above must succeed without ignoring failures.
121+
} finally {
122+
pool.close();
123+
}
124+
}
125+
}
126+
127+
@Test
128+
public void testBusinessAndBackgroundRecoveryThroughOmittedSeed() throws Exception {
129+
EnvFactory.getEnv().shutdownDataNode(0);
130+
awaitStatus(0, EnvUtils.getNodeStatusAfterLocalStop());
131+
// Control refresh timing so neither supplier learns that node 0 recovered before failover.
132+
ScheduledThreadPoolExecutor executor =
133+
new ScheduledThreadPoolExecutor(1) {
134+
@Override
135+
public ScheduledFuture<?> scheduleAtFixedRate(
136+
Runnable command, long initialDelay, long period, TimeUnit unit) {
137+
return super.scheduleAtFixedRate(command, 1, 1, TimeUnit.DAYS);
138+
}
139+
};
140+
NodesSupplier business = createSupplier(executor);
141+
NodesSupplier background = createSupplier(executor);
142+
try (Session session = new Session.Builder().nodeUrls(urls()).enableAutoFetch(false).build()) {
143+
Set<TEndPoint> discovered = new HashSet<>(Arrays.asList(endpoint(1), endpoint(2)));
144+
awaitSnapshot(business, discovered);
145+
awaitSnapshot(background, discovered);
146+
session.open(false, 1000, new ConcurrentHashMap<>(), business);
147+
148+
EnvFactory.getEnv().startDataNode(0);
149+
awaitStatus(0, NodeStatus.Running);
150+
EnvFactory.getEnv().getDataNodeWrapper(1).stopForcibly();
151+
awaitStatus(1, NodeStatus.Unknown);
152+
EnvFactory.getEnv().shutdownDataNode(2);
153+
awaitStatus(2, EnvUtils.getNodeStatusAfterLocalStop());
154+
assertFalse(business.get().contains(endpoint(0)));
155+
assertFalse(background.get().contains(endpoint(0)));
156+
157+
// Both discovered endpoints are dead. Only the initial seed omitted by discovery is live.
158+
try (SessionDataSet result = session.executeQueryStatement("SHOW AVAILABLE URLS")) {
159+
assertEquals(Collections.singleton(endpoint(0)), readEndpoints(result));
160+
}
161+
awaitSnapshot(background, Collections.singleton(endpoint(0)));
162+
assertEquals(3, background.getSeedNodes().size());
163+
// Wait for the immediate refresh requested by business reconnection, not a 60-second poll.
164+
executor.submit(() -> {}).get(10, TimeUnit.SECONDS);
165+
assertEquals(Collections.singletonList(endpoint(0)), business.get());
166+
} finally {
167+
business.close();
168+
background.close();
169+
executor.shutdownNow();
170+
assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
171+
}
172+
}
173+
174+
private void assertQuery(Session session, SessionPool pool) throws Exception {
175+
String sql = "SELECT s FROM root.seed_fallback.d";
176+
try (SessionDataSet result = session.executeQueryStatement(sql)) {
177+
assertTrue(result.hasNext());
178+
assertEquals(42, result.next().getFields().get(0).getIntV());
179+
assertFalse(result.hasNext());
180+
}
181+
try (SessionDataSetWrapper result = pool.executeQueryStatement(sql)) {
182+
assertTrue(result.hasNext());
183+
assertEquals(42, result.next().getFields().get(0).getIntV());
184+
assertFalse(result.hasNext());
185+
}
186+
}
187+
188+
private NodesSupplier supplier(Object owner) throws Exception {
189+
Field field = owner.getClass().getDeclaredField("availableNodes");
190+
field.setAccessible(true);
191+
return (NodesSupplier) field.get(owner);
192+
}
193+
194+
private NodesSupplier createSupplier(ScheduledThreadPoolExecutor executor) {
195+
List<TEndPoint> seeds = Arrays.asList(endpoint(0), endpoint(1), endpoint(2));
196+
return NodesSupplier.createNodeSupplier(
197+
seeds,
198+
executor,
199+
"root",
200+
"root",
201+
ZoneId.of("UTC"),
202+
SessionConfig.DEFAULT_INITIAL_BUFFER_CAPACITY,
203+
SessionConfig.DEFAULT_MAX_FRAME_SIZE,
204+
1000,
205+
false,
206+
null,
207+
null,
208+
null,
209+
null,
210+
"TLS",
211+
false,
212+
"V_1_0");
213+
}
214+
215+
private Set<TEndPoint> readEndpoints(SessionDataSet dataSet) throws Exception {
216+
Set<TEndPoint> result = new HashSet<>();
217+
SessionDataSet.DataIterator iterator = dataSet.iterator();
218+
while (iterator.next()) {
219+
result.add(new TEndPoint(iterator.getString("RpcAddress"), iterator.getInt("RpcPort")));
220+
}
221+
return result;
222+
}
223+
224+
private TEndPoint endpoint(int index) {
225+
return new TEndPoint(
226+
EnvFactory.getEnv().getDataNodeWrapper(index).getIp(),
227+
EnvFactory.getEnv().getDataNodeWrapper(index).getPort());
228+
}
229+
230+
private List<String> urls() {
231+
return Arrays.asList(endpoint(0), endpoint(1), endpoint(2)).stream()
232+
.map(UrlUtils::convertTEndPointIpv4AndIpv6Url)
233+
.collect(Collectors.toCollection(ArrayList::new));
234+
}
235+
236+
private void awaitStatus(int index, NodeStatus status) {
237+
EnvFactory.getEnv()
238+
.ensureNodeStatus(
239+
Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(index)),
240+
Collections.singletonList(status));
241+
}
242+
243+
private void awaitSnapshot(NodesSupplier current, Set<TEndPoint> expected) {
244+
Awaitility.await()
245+
.atMost(30, TimeUnit.SECONDS)
246+
.pollInterval(100, TimeUnit.MILLISECONDS)
247+
.untilAsserted(
248+
() -> {
249+
// run() can return immediately while a scheduled refresh is still in flight.
250+
current.run();
251+
assertEquals(expected, new HashSet<>(current.get()));
252+
});
253+
}
254+
}

‎iotdb-client/isession/src/main/java/org/apache/iotdb/isession/INodeSupplier.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
2323

24+
import java.util.Collections;
2425
import java.util.List;
2526
import java.util.Optional;
2627
import java.util.function.Supplier;
@@ -30,4 +31,14 @@ public interface INodeSupplier extends Supplier<List<TEndPoint>> {
3031
void close();
3132

3233
Optional<TEndPoint> getQueryEndPoint();
34+
35+
/** Initial addresses to try after all discovered endpoints have failed. */
36+
default List<TEndPoint> getSeedNodes() {
37+
return Collections.emptyList();
38+
}
39+
40+
/** Request a discovery refresh after reconnecting through a seed. */
41+
default void requestRefresh() {
42+
// Suppliers without automatic discovery have nothing to refresh.
43+
}
3344
}

0 commit comments

Comments
 (0)