From 7e19d4aa6771ef87946d3b14e5d94b1bf84745ad Mon Sep 17 00:00:00 2001 From: ConradJam Date: Fri, 16 May 2025 15:07:35 +0800 Subject: [PATCH 1/6] Support multiple nodes to access Amoro Rest service in a high availability environment --- .../amoro/server/AmoroServiceContainer.java | 33 +++++++++++++------ .../apache/amoro/server/AmsEnvironment.java | 4 +-- 2 files changed, 25 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 418b60a076..a343cab656 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 39384c68c2..d5a6e33e98 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 @@ -151,7 +151,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); @@ -315,7 +315,7 @@ private void startAms() throws Exception { AmoroManagementConf.OPTIMIZING_SERVICE_THRIFT_BIND_PORT, optimizingServiceBindPort); serviceConfig.set( AmoroManagementConf.REFRESH_EXTERNAL_CATALOGS_INTERVAL, Duration.ofMillis(1000L)); - serviceContainer.startService(); + serviceContainer.startOptimizingService(); LOG.info("Started test AMS."); break; } catch (TTransportException e) { From 815e96c205a343110b630b1a76b03d1fb141ddb1 Mon Sep 17 00:00:00 2001 From: zhoujinsong Date: Tue, 19 Aug 2025 17:35:02 +0800 Subject: [PATCH 2/6] Fix conflicts --- .../java/org/apache/amoro/server/AmoroServiceContainer.java | 4 ++-- 1 file changed, 2 insertions(+), 2 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 a343cab656..818694cd00 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(); From 6c90c3605d7bb09f81bd3008aa75ca4774282858 Mon Sep 17 00:00:00 2001 From: ConradJam Date: Thu, 29 May 2025 14:22:50 +0800 Subject: [PATCH 3/6] 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 818694cd00..378bcc8e4c 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 d5a6e33e98..389f2901f4 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 @@ -151,7 +151,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); From 8ece674d552fc488d0e711c50e6863c349c8418f Mon Sep 17 00:00:00 2001 From: zhoujinsong Date: Fri, 6 Jun 2025 15:19:18 +0800 Subject: [PATCH 4/6] Split dispose processing of rest service and optimizing service --- .../amoro/server/AmoroServiceContainer.java | 41 +++++++++++-------- 1 file changed, 25 insertions(+), 16 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 378bcc8e4c..3a46396b73 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 @@ -116,8 +116,9 @@ public AmoroServiceContainer() throws Exception { } public static void main(String[] args) { + AmoroServiceContainer service = null; try { - AmoroServiceContainer service = new AmoroServiceContainer(); + service = new AmoroServiceContainer(); Runtime.getRuntime() .addShutdownHook( new Thread( @@ -132,16 +133,17 @@ public static void main(String[] args) { service.waitLeaderShip(); service.startOptimizingService(); service.waitFollowerShip(); - // become follower, dispose optimizingService stop - service.stopOptimizingService(); } catch (Exception e) { LOG.error("AMS start error", e); } finally { - service.dispose(); + service.disposeOptimizingService(); } } } catch (Throwable t) { LOG.error("AMS encountered an unknown exception, will exist", t); + if (service != null) { + service.disposeRestService(); + } System.exit(1); } } @@ -209,7 +211,7 @@ private void addHandlerChain(RuntimeHandlerChain chain) { } } - public void dispose() { + public void disposeOptimizingService() { if (tableManagementServer != null && tableManagementServer.isServing()) { LOG.info("Stopping table management server..."); tableManagementServer.stop(); @@ -218,6 +220,19 @@ public void dispose() { LOG.info("Stopping optimizing server..."); optimizingServiceServer.stop(); } + if (tableService != null) { + LOG.info("Stopping table service..."); + tableService.dispose(); + tableService = null; + } + if (optimizingService != null) { + LOG.info("Stopping optimizing service..."); + optimizingService.dispose(); + optimizingService = null; + } + } + + public void disposeRestService() { if (httpServer != null) { LOG.info("Stopping http server..."); try { @@ -226,22 +241,11 @@ public void dispose() { LOG.error("Error stopping http server", e); } } - if (tableService != null) { - LOG.info("Stopping table service..."); - tableService.dispose(); - tableService = null; - } if (terminalManager != null) { LOG.info("Stopping terminal manager..."); terminalManager.dispose(); terminalManager = null; } - if (optimizingService != null) { - LOG.info("Stopping optimizing service..."); - optimizingService.dispose(); - optimizingService = null; - } - if (amsServiceMetrics != null) { amsServiceMetrics.unregister(); } @@ -250,6 +254,11 @@ public void dispose() { MetricManager.dispose(); } + public void dispose() { + disposeOptimizingService(); + disposeRestService(); + } + private void initConfig() throws Exception { LOG.info("initializing configurations..."); new ConfigurationHelper().init(); From c79a8a2ad95a0e51613b999bd62d3b39b90e2a6c Mon Sep 17 00:00:00 2001 From: zhoujinsong Date: Fri, 6 Jun 2025 15:27:15 +0800 Subject: [PATCH 5/6] as head --- .../java/org/apache/amoro/server/AmoroServiceContainer.java | 6 +----- 1 file changed, 1 insertion(+), 5 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 3a46396b73..8c5aa06d81 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 @@ -116,9 +116,8 @@ public AmoroServiceContainer() throws Exception { } public static void main(String[] args) { - AmoroServiceContainer service = null; try { - service = new AmoroServiceContainer(); + AmoroServiceContainer service = new AmoroServiceContainer(); Runtime.getRuntime() .addShutdownHook( new Thread( @@ -141,9 +140,6 @@ public static void main(String[] args) { } } catch (Throwable t) { LOG.error("AMS encountered an unknown exception, will exist", t); - if (service != null) { - service.disposeRestService(); - } System.exit(1); } } From f2c377371c27401cfc9dfaab987990a01b72c487 Mon Sep 17 00:00:00 2001 From: zhoujinsong Date: Tue, 19 Aug 2025 17:37:21 +0800 Subject: [PATCH 6/6] Remove useless codes --- .../org/apache/amoro/server/AmoroServiceContainer.java | 8 -------- .../test/java/org/apache/amoro/server/AmsEnvironment.java | 1 + 2 files changed, 1 insertion(+), 8 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 8c5aa06d81..4d418d7f52 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 @@ -144,14 +144,6 @@ 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(); } 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 389f2901f4..827135da42 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 @@ -315,6 +315,7 @@ private void startAms() throws Exception { AmoroManagementConf.OPTIMIZING_SERVICE_THRIFT_BIND_PORT, optimizingServiceBindPort); serviceConfig.set( AmoroManagementConf.REFRESH_EXTERNAL_CATALOGS_INTERVAL, Duration.ofMillis(1000L)); + serviceContainer.startRestServices(); serviceContainer.startOptimizingService(); LOG.info("Started test AMS."); break;