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 <[email protected]> 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.
