You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/05/10 14:21:43 UTC
[4/9] flink git commit: [FLINK-9190][flip6,
yarn] Request new container if container completed unexpectedly.
[FLINK-9190][flip6,yarn] Request new container if container completed unexpectedly.
This closes #5931.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fa11015d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fa11015d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fa11015d
Branch: refs/heads/master
Commit: fa11015ddcdb0217da6a8b2be1e2a55efd26d7fa
Parents: b7f5c0d
Author: gyao <ga...@data-artisans.com>
Authored: Fri Apr 27 15:49:31 2018 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu May 10 16:18:31 2018 +0200
----------------------------------------------------------------------
.../apache/flink/yarn/YarnResourceManager.java | 44 ++++++++++----------
1 file changed, 23 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/fa11015d/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
index 22d0164..266c037 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -295,21 +295,16 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
}
@Override
- public boolean stopWorker(YarnWorkerNode workerNode) {
- if (workerNode != null) {
- Container container = workerNode.getContainer();
- log.info("Stopping container {}.", container.getId());
- // release the container on the node manager
- try {
- nodeManagerClient.stopContainer(container.getId(), container.getNodeId());
- } catch (Throwable t) {
- log.warn("Error while calling YARN Node Manager to stop container", t);
- }
- resourceManagerClient.releaseAssignedContainer(container.getId());
- workerNodeMap.remove(workerNode.getResourceID());
- } else {
- log.error("Can not find container for null workerNode.");
+ public boolean stopWorker(final YarnWorkerNode workerNode) {
+ final Container container = workerNode.getContainer();
+ log.info("Stopping container {}.", container.getId());
+ try {
+ nodeManagerClient.stopContainer(container.getId(), container.getNodeId());
+ } catch (final Exception e) {
+ log.warn("Error while calling YARN Node Manager to stop container", e);
}
+ resourceManagerClient.releaseAssignedContainer(container.getId());
+ workerNodeMap.remove(workerNode.getResourceID());
return true;
}
@@ -329,14 +324,19 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
}
@Override
- public void onContainersCompleted(List<ContainerStatus> list) {
+ public void onContainersCompleted(final List<ContainerStatus> list) {
runAsync(() -> {
- for (ContainerStatus container : list) {
- if (container.getExitStatus() < 0) {
- closeTaskManagerConnection(new ResourceID(
- container.getContainerId().toString()), new Exception(container.getDiagnostics()));
+ for (final ContainerStatus containerStatus : list) {
+
+ final ResourceID resourceId = new ResourceID(containerStatus.getContainerId().toString());
+ final YarnWorkerNode yarnWorkerNode = workerNodeMap.remove(resourceId);
+
+ if (yarnWorkerNode != null) {
+ // Container completed unexpectedly ~> start a new one
+ final Container container = yarnWorkerNode.getContainer();
+ requestYarnContainer(container.getResource(), yarnWorkerNode.getContainer().getPriority());
+ closeTaskManagerConnection(resourceId, new Exception(containerStatus.getDiagnostics()));
}
- workerNodeMap.remove(new ResourceID(container.getContainerId().toString()));
}
}
);
@@ -355,8 +355,9 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
numPendingContainerRequests--;
final String containerIdStr = container.getId().toString();
+ final ResourceID resourceId = new ResourceID(containerIdStr);
- workerNodeMap.put(new ResourceID(containerIdStr), new YarnWorkerNode(container));
+ workerNodeMap.put(resourceId, new YarnWorkerNode(container));
try {
// Context information used to start a TaskExecutor Java process
@@ -370,6 +371,7 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
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
requestYarnContainer(container.getResource(), container.getPriority());