You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2022/10/21 12:32:39 UTC
[iotdb] 02/02: fix some bugs
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch IOTDB-4619
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 186452c4973c7649e7075df501851ae6c58ead0e
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Fri Oct 21 20:32:28 2022 +0800
fix some bugs
---
.../iotdb/confignode/manager/cq/CQManager.java | 8 +++++--
.../iotdb/confignode/manager/node/NodeManager.java | 27 ++++++++--------------
.../org/apache/iotdb/it/env/RemoteServerEnv.java | 13 ++++++++++-
.../confignode/it/IoTDBConfigNodeSnapshotIT.java | 8 +++----
4 files changed, 32 insertions(+), 24 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQManager.java
index b0995f4496..90e8a32a58 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQManager.java
@@ -117,7 +117,9 @@ public class CQManager {
try {
// 1. shutdown previous cq schedule thread pool
try {
- executor.shutdown();
+ if (executor != null) {
+ executor.shutdown();
+ }
} catch (Throwable t) {
// just print the error log because we should make sure we can start a new cq schedule pool
// successfully in the next steps
@@ -166,6 +168,8 @@ public class CQManager {
} finally {
lock.writeLock().unlock();
}
- previous.shutdown();
+ if (previous != null) {
+ previous.shutdown();
+ }
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index 875017c8f2..844971242e 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -85,6 +85,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
@@ -133,12 +134,15 @@ public class NodeManager {
IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor("Unknown-DataNode-Detector");
private final Set<TDataNodeLocation> oldUnknownNodes;
+ private final Random random;
+
public NodeManager(IManager configManager, NodeInfo nodeInfo) {
this.configManager = configManager;
this.nodeInfo = nodeInfo;
this.removeConfigNodeLock = new ReentrantLock();
this.nodeCacheMap = new ConcurrentHashMap<>();
this.oldUnknownNodes = new HashSet<>();
+ this.random = new Random(System.currentTimeMillis());
}
private void setGlobalConfig(DataNodeRegisterResp dataSet) {
@@ -829,26 +833,15 @@ public class NodeManager {
* @return TDataNodeLocation
*/
public Optional<TDataNodeLocation> getLowestLoadDataNode() {
- AtomicInteger result = new AtomicInteger(-1);
- AtomicLong lowestLoadScore = new AtomicLong(Long.MAX_VALUE);
-
- nodeCacheMap.forEach(
- (dataNodeId, heartbeatCache) -> {
- long score = heartbeatCache.getLoadScore();
- if (heartbeatCache.getNodeStatus() == NodeStatus.Running
- && score < lowestLoadScore.get()) {
- result.set(dataNodeId);
- lowestLoadScore.set(score);
- }
- });
+ // TODO get real lowest load data node after scoring algorithm being implemented
+ List<TDataNodeConfiguration> targetDataNodeList =
+ filterDataNodeThroughStatus(NodeStatus.Running);
- if (result.get() == -1) {
+ if (targetDataNodeList == null || targetDataNodeList.isEmpty()) {
return Optional.empty();
} else {
- LOGGER.info(
- "get the lowest load DataNode, NodeID: [{}], LoadScore: [{}]", result, lowestLoadScore);
- return Optional.of(
- configManager.getNodeManager().getRegisteredDataNodeLocations().get(result.get()));
+ int index = random.nextInt(targetDataNodeList.size());
+ return Optional.of(targetDataNodeList.get(index).location);
}
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/RemoteServerEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/RemoteServerEnv.java
index 95cf12bf35..438989e432 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/RemoteServerEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/RemoteServerEnv.java
@@ -18,7 +18,11 @@
*/
package org.apache.iotdb.it.env;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService;
+import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
import org.apache.iotdb.itbase.env.BaseEnv;
import org.apache.iotdb.jdbc.Config;
import org.apache.iotdb.jdbc.Constant;
@@ -148,7 +152,14 @@ public class RemoteServerEnv implements BaseEnv {
@Override
public IConfigNodeRPCService.Iface getLeaderConfigNodeConnection() throws IOException {
- return null;
+ IClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager =
+ new IClientManager.Factory<TEndPoint, SyncConfigNodeIServiceClient>()
+ .createClientManager(
+ new DataNodeClientPoolFactory.SyncConfigNodeIServiceClientPoolFactory());
+ try (SyncConfigNodeIServiceClient client =
+ clientManager.borrowClient(new TEndPoint(ip_addr, 22277))) {
+ return client;
+ }
}
@Override
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java
index 8c38aa1c7e..5898c817f2 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java
@@ -291,8 +291,8 @@ public class IoTDBConfigNodeSnapshotIT {
}
private Set<TCQEntry> createCQs(SyncConfigNodeIServiceClient client) throws TException {
- String sql1 = "create cq testCq1 BEGIN select s1 into root.backup.d1.s1 from root.sg.d1 END";
- String sql2 = "create cq testCq1 BEGIN select s1 into root.backup.d2.s1 from root.sg.d2 END";
+ String sql1 = "create cq testCq1 BEGIN select s1 into root.backup.d1(s1) from root.sg.d1 END";
+ String sql2 = "create cq testCq2 BEGIN select s1 into root.backup.d2(s1) from root.sg.d2 END";
TCreateCQReq req1 =
new TCreateCQReq(
"testCq1",
@@ -301,7 +301,7 @@ public class IoTDBConfigNodeSnapshotIT {
1000,
0,
(byte) 0,
- "select s1 into root.backup.d1.s1 from root.sg.d1",
+ "select s1 into root.backup.d1(s1) from root.sg.d1",
sql1,
"Asia");
TCreateCQReq req2 =
@@ -312,7 +312,7 @@ public class IoTDBConfigNodeSnapshotIT {
1000,
0,
(byte) 1,
- "select s1 into root.backup.d2.s1 from root.sg.d2",
+ "select s1 into root.backup.d2(s1) from root.sg.d2",
sql2,
"Asia");