This is an automated email from the ASF dual-hosted git repository. haonan pushed a commit to branch IOTDB-3306-13 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ae4dea386acf3759e7496dfb06c1c3d967a9b89b Author: HTHou <[email protected]> AuthorDate: Mon May 30 18:07:32 2022 +0800 [To rel/0.13][IOTDB-3306] Use rpc port check to avoid starting same IoTDB twice --- .../db/engine/storagegroup/TsFileProcessor.java | 2 +- .../java/org/apache/iotdb/db/service/IoTDB.java | 9 +++-- .../db/service/thrift/ThriftServiceThread.java | 38 ++-------------------- 3 files changed, 8 insertions(+), 41 deletions(-) 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 187081f72d..4a9f9043a0 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 @@ -1031,7 +1031,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/IoTDB.java b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java index 501acdc4cd..58b850f71c 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 @@ -134,6 +134,10 @@ public class IoTDB implements IoTDBMBean { Runtime.getRuntime().addShutdownHook(new IoTDBShutdownHook()); setUncaughtExceptionHandler(); + // 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..."); initMManager(); @@ -149,11 +153,6 @@ public class IoTDB implements IoTDBMBean { registerManager.register(UDFClassLoaderManager.getInstance()); registerManager.register(UDFRegistrationService.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. diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/ThriftServiceThread.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/ThriftServiceThread.java index 462a2e5d14..53cf755ad0 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/thrift/ThriftServiceThread.java +++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/ThriftServiceThread.java @@ -222,45 +222,13 @@ public class ThriftServiceThread 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) {
