Skip to content

[Refactor] Unify the main process of amoro service container #4339

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

Merged
czy006 merged 5 commits into from
Aug 31, 2026
+90 −62
Merged

[Refactor] Unify the main process of amoro service container #4339

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0b00a6b
Unify the main process of amoro service container
zhoujinsong Aug 19, 2026
0e672e2
Fix checkstyle issus
zhoujinsong Aug 26, 2026
fc6e75e
Merge branch 'master' into refactor-master-slave-mode
zhoujinsong Aug 26, 2026
0c75541
Merge branch 'master' into refactor-master-slave-mode
zhoujinsong Aug 26, 2026
ab23b1e
Stop leader service first in dispose path
zhoujinsong Aug 28, 2026
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
148 changes: 88 additions & 60 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 @@ -131,6 +131,7 @@ public class AmoroServiceContainer {
public AmoroServiceContainer() throws Exception {
initConfig();
haContainer = HighAvailabilityContainerFactory.create(serviceConfig);
haContainer.registerAndElect();
}

public static void main(String[] args) {
Expand All @@ -144,28 +145,21 @@ public static void main(String[] args) {
service.dispose();
LOG.info("AMS service has been shut down");
}));
service.startRestServices();
if (IS_MASTER_SLAVE_MODE) {
// Even if one does not become the master, it cannot block the subsequent logic.
service.registAndElect();
// Regardless of whether tp becomes the master, the service needs to be activated.
service.startOptimizingService();
} else {
while (true) {
try {
// Used to block AMS instances that have not acquired leadership
service.waitLeaderShip();
service.transitionToLeader();
// Used to block AMS instances that have acquired leadership
service.waitFollowerShip();
} catch (ConfigurationException e) {
LOG.error("AMS will exit...", e);
System.exit(1);
} catch (Exception e) {
LOG.error("AMS start error", e);
} finally {
service.transitionToFollower();
}

service.startBaseServices();

while (true) {
try {
service.waitLeaderShip();
service.startLeaderServices();
service.waitFollowerShip();
} catch (ConfigurationException e) {
LOG.error("AMS will exit...", e);
System.exit(1);
} catch (Exception e) {
LOG.error("AMS start error", e);
} finally {
service.stopLeaderServices();
}
}
} catch (Throwable t) {
Expand All @@ -174,10 +168,6 @@ public static void main(String[] args) {
}
}

public void registAndElect() throws Exception {
haContainer.registerAndElect();
}

public enum HAState {
INITIALIZING(0),
FOLLOWER(1),
Expand Down Expand Up @@ -220,24 +210,25 @@ public void startRestServices() throws Exception {
registerAmsServiceMetric();
}

public void transitionToLeader() throws Exception {
if (haState == HAState.LEADER) {
return;
}
startOptimizingService();
haState = HAState.LEADER;
}

public void transitionToFollower() {
if (haState == HAState.FOLLOWER) {
return;
/**
* Start base services that every AMS node needs regardless of HA mode: REST/HTTP and
* catalog/table managers. In master-slave mode, also starts the optimizing service (including
* Thrift), because non-leader nodes must serve optimizer requests. In active-standby mode, the
* optimizing service is deferred to {@link #startLeaderServices} since the standby node does not
* serve any requests.
*/
public void startBaseServices() throws Exception {
startRestServices();
if (IS_MASTER_SLAVE_MODE) {
startOptimizingService();
}
haState = HAState.FOLLOWER;
disposeOptimizingService();
}

public void startOptimizingService() throws Exception {

/**
* Create optimizing service objects, register handler chains, initialize table service, and start
* the Thrift servers.
*/
private void startOptimizingService() throws Exception {
// Load process factories and build action coordinators from default table runtime factory.
TableProcessFactoryManager tableProcessFactoryManager = new TableProcessFactoryManager();
tableProcessFactoryManager.initialize();
Expand All @@ -255,20 +246,6 @@ public void startOptimizingService() throws Exception {
bucketAssignStore = BucketAssignStoreFactory.create(haContainer, serviceConfig);
}

// In master-slave mode, create AmsAssignService for bucket assignment (shares BucketAssignStore
// with DefaultTableService).
if (IS_MASTER_SLAVE_MODE && haContainer != null && bucketAssignStore != null) {
try {
amsAssignService = new AmsAssignService(haContainer, serviceConfig, bucketAssignStore);
amsAssignService.start();
LOG.info("AmsAssignService started for master-slave mode");
} catch (UnsupportedOperationException e) {
LOG.info("Skip AmsAssignService: {}", e.getMessage());
} catch (Exception e) {
LOG.error("Failed to start AmsAssignService", e);
}
}

List<ActionCoordinator> actionCoordinators = defaultRuntimeFactory.supportedCoordinators();

tableService =
Expand Down Expand Up @@ -300,6 +277,61 @@ public void startOptimizingService() throws Exception {
startThriftService();
}

/**
* Start leader-exclusive services. In active-standby mode this starts the optimizing service
* (including Thrift) since only the leader serves requests. In master-slave mode the optimizing
* service is already started in {@link #startBaseServices} because non-leader nodes also serve
* optimizer requests; here only the leader-exclusive schedulers (e.g. AmsAssignService) are
* started.
*/
public void startLeaderServices() throws Exception {
if (haState == HAState.LEADER) {
return;
}
if (IS_MASTER_SLAVE_MODE) {
// AmsAssignService may have been stopped and set to null by a previous stopLeaderServices
// call (leader re-election); recreate it if needed.
if (amsAssignService == null && haContainer != null) {
try {
BucketAssignStore bucketAssignStore =
BucketAssignStoreFactory.create(haContainer, serviceConfig);
amsAssignService = new AmsAssignService(haContainer, serviceConfig, bucketAssignStore);
} catch (Exception e) {
LOG.error("Failed to recreate AmsAssignService", e);
}
}
if (amsAssignService != null) {
amsAssignService.start();
LOG.info("AmsAssignService started");
}
} else {
startOptimizingService();
}
haState = HAState.LEADER;
}

/**
* Stop leader-exclusive services. In active-standby mode this disposes the entire optimizing
* service (including Thrift and service objects), since the standby node does not serve any
* requests. In master-slave mode only the leader-exclusive schedulers are stopped; Thrift stays
* running so the node can continue serving optimizer requests.
*/
public void stopLeaderServices() {
if (haState == HAState.FOLLOWER) {
return;
}
if (IS_MASTER_SLAVE_MODE) {
if (amsAssignService != null) {
LOG.info("Stopping AmsAssignService...");
amsAssignService.stop();
amsAssignService = null;
}
} else {
disposeOptimizingService();
}
haState = HAState.FOLLOWER;
}

private void addHandlerChain(RuntimeHandlerChain chain) {
if (chain != null) {
tableService.addHandlerChain(chain);
Expand All @@ -316,11 +348,6 @@ public void disposeOptimizingService() {
LOG.info("Stopping optimizing server[serving:{}] ...", optimizingServiceServer.isServing());
optimizingServiceServer.stop();
}
if (amsAssignService != null) {

@xxubai xxubai Aug 28, 2026

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.

dispose() still calls disposeOptimizingService() directly from the shutdown hook. Will it cause the ams-assign-scheduler running while the other services are being disposed?

@zhoujinsong zhoujinsong Aug 28, 2026 •
edited
Loading

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.

Yes, we may stop the leader service first, if necessary, when disposing.
I have made some changes, PTAL.

LOG.info("Stopping AmsAssignService...");
amsAssignService.stop();
amsAssignService = null;
}
if (tableService != null) {
LOG.info("Stopping table service...");
tableService.dispose();
Expand Down Expand Up @@ -360,6 +387,7 @@ public void disposeRestService() {
}

public void dispose() {
stopLeaderServices();
disposeOptimizingService();
disposeRestService();
}
Expand Down
4 changes: 2 additions & 2 deletions 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 @@ -333,8 +333,8 @@ 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();
serviceContainer.startBaseServices();
serviceContainer.startLeaderServices();
LOG.info("Started test AMS.");
break;
} catch (TTransportException e) {
Expand Down
Loading