xintongsong commented on a change in pull request #9106: [FLINK-13184][yarn] 
Support launching task executors with multi-thread on YARN.
URL: https://github.com/apache/flink/pull/9106#discussion_r313764027
 
 

 ##########
 File path: 
flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
 ##########
 @@ -391,23 +416,33 @@ public void onContainersAllocated(List<Container> 
containers) {
 
                                        workerNodeMap.put(resourceId, new 
YarnWorkerNode(container));
 
-                                       try {
-                                               // Context information used to 
start a TaskExecutor Java process
-                                               ContainerLaunchContext 
taskExecutorLaunchContext = createTaskExecutorLaunchContext(
-                                                       container.getResource(),
-                                                       containerIdStr,
-                                                       
container.getNodeId().getHost());
-
-                                               
nodeManagerClient.startContainer(container, taskExecutorLaunchContext);
-                                       } catch (Throwable t) {
-                                               log.error("Could not start 
TaskManager in container {}.", container.getId(), t);
-
-                                               // release the failed container
-                                               
workerNodeMap.remove(resourceId);
-                                               
resourceManagerClient.releaseAssignedContainer(container.getId());
-                                               // and ask for a new one
-                                               
requestYarnContainerIfRequired();
-                                       }
+                                       startContainerExecutor.execute(new 
Runnable() {
+                                               @Override
+                                               public void run() {
+                                                       if 
(workerNodeMap.get(resourceId) == null) {
+                                                               log.info("Skip 
launching container {}, container doesn't exist in container worker map",
+                                                                       
containerIdStr);
+                                                               return;
+                                                       }
+                                                       try {
+                                                               // Context 
information used to start a TaskExecutor Java process
+                                                               
ContainerLaunchContext taskExecutorLaunchContext = 
createTaskExecutorLaunchContext(
+                                                                       
container.getResource(),
+                                                                       
containerIdStr,
+                                                                       
container.getNodeId().getHost());
+
+                                                               
nodeManagerClient.startContainer(container, taskExecutorLaunchContext);
 
 Review comment:
   Yes, it is. I checked the source codes myself, and also double checked with 
Tao Yang who is an apache hadoop committer.

----------------------------------------------------------------
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


With regards,
Apache Git Services

Reply via email to