Skip to content

[AMORO-2810]: Support multiple nodes to access Amoro Rest service in a high availability environment #3567

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
czy006 wants to merge 8 commits into from
Closed

[AMORO-2810]: Support multiple nodes to access Amoro Rest service in a high availability environment #3567

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
906037f
Support multiple nodes to access Amoro Rest service in a high availab…
czy006 May 16, 2025
9b9ab71
Merge branch 'master' into zk-ha-rest
czy006 May 22, 2025
32ebe52
Merge branch 'master' into zk-ha-rest
czy006 May 23, 2025
2038dcc
rename to startRestServices
czy006 May 26, 2025
0758f38
Merge branch 'master' into zk-ha-rest
xxubai May 29, 2025
656a349
server dispose rename
czy006 May 29, 2025
b3c26c6
Merge branch 'master' into zk-ha-rest
czy006 May 29, 2025
55b63e4
Merge branch 'master' into zk-ha-rest
zhoujinsong Jun 6, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 20 additions & 7 deletions amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,11 +126,14 @@ public static void main(String[] args) {
service.dispose();
LOG.info("AMS service has been shut down");
}));
service.startRestServices();
while (true) {
try {
service.waitLeaderShip();
service.startService();
service.startOptimizingService();
service.waitFollowerShip();
// become follower, dispose optimizingService stop
service.stopOptimizingService();

@Aireed Aireed May 29, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why didn't you put this in service.dispose? or you may change service.dispose in finally block to service.stopOptimizingService ???

@czy006 czy006 May 29, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

dispose will be close all service (include rest) .if this node become follower,we just close optimizingService,and will be run waitting become Leader service.waitLeaderShip()

@Aireed Aireed May 29, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But the finally block will dispose all services. so you should use stopOptimizingService to repalce dsipose in finally block ?

@zhoujinsong zhoujinsong Jun 6, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, the dispose process will close the rest service. It should be fixed.

I open a PR to fix it: czy006#2
@czy006 You can checkout it.

@czy006 czy006 Jun 10, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the feedback. I am using the new logic to verify it, which will take some time. At the same time, we are designing a more general high-availability solution, which will support K8S in the future and will be launched later. cc @Aireed @zhoujinsong @klion26

@zhoujinsong zhoujinsong Jun 10, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds great! Look forward it!

} catch (Exception e) {
LOG.error("AMS start error", e);
} finally {
Expand All @@ -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();
}
Expand All @@ -151,14 +162,21 @@ public void waitFollowerShip() throws Exception {
haContainer.waitFollowerShip();
}

public void startService() throws Exception {
public void startRestServices() 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 =
Expand All @@ -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) {
Expand Down
3 changes: 2 additions & 1 deletion amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -349,7 +349,8 @@ private void startAms() throws Exception {
serviceConfig.set(
AmoroManagementConf.REFRESH_EXTERNAL_CATALOGS_INTERVAL,
Duration.ofMillis(1000L));
serviceContainer.startService();
serviceContainer.startRestServices();
serviceContainer.startOptimizingService();
break;
} catch (TTransportException e) {
if (e.getCause() instanceof BindException) {
Expand Down