From 906037f4014d952229c90b2f0bc903752525cab7 Mon Sep 17 00:00:00 2001 From: ConradJam Date: Fri, 16 May 2025 15:07:35 +0800 Subject: [PATCH 1/3] Support multiple nodes to access Amoro Rest service in a high availability environment --- .../amoro/server/AmoroServiceContainer.java | 33 +++++++++++++------ .../apache/amoro/server/AmsEnvironment.java | 5 +-- 2 files changed, 26 insertions(+), 12 deletions(-) diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java index 283d211dd8..d502fd5dd7 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java @@ -123,18 +123,21 @@ public static void main(String[] args) { new Thread( () -> { LOG.info("AMS service is shutting down..."); - service.dispose(); + service.disposeAllService(); LOG.info("AMS service has been shut down"); })); + service.startNoHighAvailableService(); while (true) { try { service.waitLeaderShip(); - service.startService(); + service.startOptimizingService(); service.waitFollowerShip(); + // become follower, dispose optimizingService stop + service.stopOptimizingService(); } catch (Exception e) { LOG.error("AMS start error", e); } finally { - service.dispose(); + service.disposeAllService(); } } } catch (Throwable t) { @@ -143,6 +146,14 @@ public static void main(String[] args) { } } + private void stopOptimizingService() { + if (optimizingService != null) { + LOG.info("AMS Become follower and Stopping optimizing service..."); + optimizingService.dispose(); + optimizingService = null; + } + } + public void waitLeaderShip() throws Exception { haContainer.waitLeaderShip(); } @@ -151,14 +162,21 @@ public void waitFollowerShip() throws Exception { haContainer.waitFollowerShip(); } - public void startService() throws Exception { + public void startNoHighAvailableService() throws Exception { EventsManager.getInstance(); MetricManager.getInstance(); catalogManager = new DefaultCatalogManager(serviceConfig); tableManager = new DefaultTableManager(serviceConfig, catalogManager); optimizerManager = new DefaultOptimizerManager(serviceConfig, catalogManager); + terminalManager = new TerminalManager(serviceConfig, catalogManager); + initHttpService(); + startHttpService(); + registerAmsServiceMetric(); + } + + public void startOptimizingService() throws Exception { tableService = new DefaultTableService(serviceConfig, catalogManager); optimizingService = @@ -180,14 +198,9 @@ public void startService() throws Exception { tableService.initialize(); LOG.info("AMS table service have been initialized"); tableManager.setTableService(tableService); - terminalManager = new TerminalManager(serviceConfig, catalogManager); initThriftService(); startThriftService(); - - initHttpService(); - startHttpService(); - registerAmsServiceMetric(); } private void addHandlerChain(RuntimeHandlerChain chain) { @@ -196,7 +209,7 @@ private void addHandlerChain(RuntimeHandlerChain chain) { } } - public void dispose() { + public void disposeAllService() { if (tableManagementServer != null && tableManagementServer.isServing()) { LOG.info("Stopping table management server..."); tableManagementServer.stop(); diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java index 1e37a071c9..b37d0145b1 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java @@ -176,7 +176,7 @@ public void stop() throws IOException { stopOptimizer(); if (this.serviceContainer != null) { - this.serviceContainer.dispose(); + this.serviceContainer.disposeAllService(); } testHMS.stop(); MoreFiles.deleteRecursively(Paths.get(rootPath), RecursiveDeleteOption.ALLOW_INSECURE); @@ -349,7 +349,8 @@ private void startAms() throws Exception { serviceConfig.set( AmoroManagementConf.REFRESH_EXTERNAL_CATALOGS_INTERVAL, Duration.ofMillis(1000L)); - serviceContainer.startService(); + serviceContainer.startNoHighAvailableService(); + serviceContainer.startOptimizingService(); break; } catch (TTransportException e) { if (e.getCause() instanceof BindException) { From 2038dccde9f5b4d0f75ee4d18be0bb276196c970 Mon Sep 17 00:00:00 2001 From: ConradJam Date: Mon, 26 May 2025 14:55:30 +0800 Subject: [PATCH 2/3] rename to startRestServices --- .../java/org/apache/amoro/server/AmoroServiceContainer.java | 4 ++-- .../src/test/java/org/apache/amoro/server/AmsEnvironment.java | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java index d502fd5dd7..2e3e5fa250 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java @@ -126,7 +126,7 @@ public static void main(String[] args) { service.disposeAllService(); LOG.info("AMS service has been shut down"); })); - service.startNoHighAvailableService(); + service.startRestServices(); while (true) { try { service.waitLeaderShip(); @@ -162,7 +162,7 @@ public void waitFollowerShip() throws Exception { haContainer.waitFollowerShip(); } - public void startNoHighAvailableService() throws Exception { + public void startRestServices() throws Exception { EventsManager.getInstance(); MetricManager.getInstance(); diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java index b37d0145b1..bb50671fea 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java @@ -349,7 +349,7 @@ private void startAms() throws Exception { serviceConfig.set( AmoroManagementConf.REFRESH_EXTERNAL_CATALOGS_INTERVAL, Duration.ofMillis(1000L)); - serviceContainer.startNoHighAvailableService(); + serviceContainer.startRestServices(); serviceContainer.startOptimizingService(); break; } catch (TTransportException e) { From 656a34939ecea8c985d1bd31f60d640ba32d737d Mon Sep 17 00:00:00 2001 From: ConradJam Date: Thu, 29 May 2025 14:22:50 +0800 Subject: [PATCH 3/3] server dispose rename --- .../java/org/apache/amoro/server/AmoroServiceContainer.java | 6 +++--- .../test/java/org/apache/amoro/server/AmsEnvironment.java | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java index fec7fab529..c9d205a077 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java @@ -123,7 +123,7 @@ public static void main(String[] args) { new Thread( () -> { LOG.info("AMS service is shutting down..."); - service.disposeAllService(); + service.dispose(); LOG.info("AMS service has been shut down"); })); service.startRestServices(); @@ -137,7 +137,7 @@ public static void main(String[] args) { } catch (Exception e) { LOG.error("AMS start error", e); } finally { - service.disposeAllService(); + service.dispose(); } } } catch (Throwable t) { @@ -209,7 +209,7 @@ private void addHandlerChain(RuntimeHandlerChain chain) { } } - public void disposeAllService() { + public void dispose() { if (tableManagementServer != null && tableManagementServer.isServing()) { LOG.info("Stopping table management server..."); tableManagementServer.stop(); diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java index e64723bc35..9102929b30 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java @@ -176,7 +176,7 @@ public void stop() throws IOException { stopOptimizer(); if (this.serviceContainer != null) { - this.serviceContainer.disposeAllService(); + this.serviceContainer.dispose(); } testHMS.stop(); MoreFiles.deleteRecursively(Paths.get(rootPath), RecursiveDeleteOption.ALLOW_INSECURE);