You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2022/05/30 08:17:49 UTC
[iotdb] 01/01: [IOTDB-3306] Use rpc port check to avoid starting same IoTDB twice
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch IOTDB-3306
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 72944be747dd3354871b2b1d960132c518beb0a7
Author: HTHou <hh...@outlook.com>
AuthorDate: Mon May 30 16:17:26 2022 +0800
[IOTDB-3306] Use rpc port check to avoid starting same IoTDB twice
---
.../service/AbstractThriftServiceThread.java | 38 ++--------------------
.../db/engine/storagegroup/TsFileProcessor.java | 2 +-
.../java/org/apache/iotdb/db/service/DataNode.java | 9 +++--
.../java/org/apache/iotdb/db/service/IoTDB.java | 9 +++--
4 files changed, 12 insertions(+), 46 deletions(-)
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java b/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java
index 4940c191d9..3ffadfc07f 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java
@@ -223,45 +223,13 @@ public abstract class AbstractThriftServiceThread extends Thread {
@SuppressWarnings("java:S2259")
private TServerTransport openTransport(String bindAddress, int port) throws TTransportException {
- int maxRetry = 5;
- long retryIntervalMS = 5000;
- TTransportException lastExp = null;
- for (int i = 0; i < maxRetry; i++) {
- try {
- return new TServerSocket(new InetSocketAddress(bindAddress, port));
- } catch (TTransportException e) {
- lastExp = e;
- try {
- Thread.sleep(retryIntervalMS);
- } catch (InterruptedException interruptedException) {
- Thread.currentThread().interrupt();
- break;
- }
- }
- }
- throw lastExp == null ? new TTransportException() : lastExp;
+ return new TServerSocket(new InetSocketAddress(bindAddress, port));
}
private TServerTransport openNonblockingTransport(
String bindAddress, int port, int connectionTimeoutInMS) throws TTransportException {
- int maxRetry = 5;
- long retryIntervalMS = 5000;
- TTransportException lastExp = null;
- for (int i = 0; i < maxRetry; i++) {
- try {
- return new TNonblockingServerSocket(
- new InetSocketAddress(bindAddress, port), connectionTimeoutInMS);
- } catch (TTransportException e) {
- lastExp = e;
- try {
- Thread.sleep(retryIntervalMS);
- } catch (InterruptedException interruptedException) {
- Thread.currentThread().interrupt();
- break;
- }
- }
- }
- throw lastExp == null ? new TTransportException() : lastExp;
+ return new TNonblockingServerSocket(
+ new InetSocketAddress(bindAddress, port), connectionTimeoutInMS);
}
public void setThreadStopLatch(CountDownLatch threadStopLatch) {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 37d67e1f3b..621eba790b 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -1222,7 +1222,7 @@ public class TsFileProcessor {
MemTableFlushTask flushTask =
new MemTableFlushTask(memTableToFlush, writer, storageGroupName);
flushTask.syncFlushMemTable();
- } catch (Exception e) {
+ } catch (Throwable e) {
if (writer == null) {
logger.info(
"{}: {} is closed during flush, abandon flush task",
diff --git a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
index e2b62e687d..cd4ca1ccad 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -266,6 +266,10 @@ public class DataNode implements DataNodeMBean {
Runtime.getRuntime().addShutdownHook(new IoTDBShutdownHook());
setUncaughtExceptionHandler();
initServiceProvider();
+ // in cluster mode, RPC service is not enabled.
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableRpcService()) {
+ registerManager.register(RPCService.getInstance());
+ }
registerManager.register(MetricsService.getInstance());
logger.info("recover the schema...");
initConfigManager();
@@ -289,11 +293,6 @@ public class DataNode implements DataNodeMBean {
registerManager.register(ReceiverService.getInstance());
registerManager.register(MetricsService.getInstance());
- // in cluster mode, RPC service is not enabled.
- if (IoTDBDescriptor.getInstance().getConfig().isEnableRpcService()) {
- registerManager.register(RPCService.getInstance());
- }
-
initProtocols();
if (IoTDBDescriptor.getInstance().getConfig().isEnableInfluxDBRpcService()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
index 5ec6baf6a6..18911877ae 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
@@ -145,6 +145,10 @@ public class IoTDB implements IoTDBMBean {
Runtime.getRuntime().addShutdownHook(new IoTDBShutdownHook());
setUncaughtExceptionHandler();
initServiceProvider();
+ // in cluster mode, RPC service is not enabled.
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableRpcService()) {
+ registerManager.register(RPCService.getInstance());
+ }
registerManager.register(MetricsService.getInstance());
logger.info("recover the schema...");
initConfigManager();
@@ -182,11 +186,6 @@ public class IoTDB implements IoTDBMBean {
registerManager.register(ReceiverService.getInstance());
registerManager.register(MetricsService.getInstance());
- // in cluster mode, RPC service is not enabled.
- if (IoTDBDescriptor.getInstance().getConfig().isEnableRpcService()) {
- registerManager.register(RPCService.getInstance());
- }
-
initProtocols();
// in cluster mode, InfluxDBMManager has been initialized, so there is no need to init again to
// avoid wasting time.