lakshmi-manasa-g commented on a change in pull request #1448:
URL: https://github.com/apache/samza/pull/1448#discussion_r537659102
##########
File path:
samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java
##########
@@ -236,11 +236,20 @@ public void start() {
diagnosticsManager.get().start();
}
+ if (jobConfig.getApplicationMasterHighAvailabilityEnabled()) {
+ LOG.info(
+ "Set neededProcessors prior to starting clusterResourceManager
because it gets running containres from prev attempts in AM HA.");
+
state.processorCount.set(state.jobModelManager.jobModel().getContainers().size());
+
state.neededProcessors.set(state.jobModelManager.jobModel().getContainers().size());
+ }
+
LOG.info("Starting the cluster resource manager");
clusterResourceManager.start();
-
state.processorCount.set(state.jobModelManager.jobModel().getContainers().size());
-
state.neededProcessors.set(state.jobModelManager.jobModel().getContainers().size());
+ if (!jobConfig.getApplicationMasterHighAvailabilityEnabled()) {
+
state.processorCount.set(state.jobModelManager.jobModel().getContainers().size());
+
state.neededProcessors.set(state.jobModelManager.jobModel().getContainers().size());
+ }
Review comment:
(i had written the response but forgot to submit it.)
the reason for this flow is to keep the non-AMHA exactly the same as before.
the big difference is setting neededProcessors prior to
ClusterResourceManager.start() as it will invoke launch sucess which inturn
will decrement neededProcessors and set job healthy accordingly.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
