Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6192#discussion_r198917467
--- Diff:
flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java ---
@@ -334,8 +335,11 @@ public void onContainersCompleted(final
List<ContainerStatus> list) {
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()));
+ // check WorkerRegistration
status to avoid requesting containers more than required
+ if
(checkWorkerRegistrationWithResourceId(resourceId)) {
--- End diff --
Wouldn't that prevent container restarts if the container failure happened
before the `TaskManager` registered at the `ResourceManager`, because then,
`ResourceManager#taskExecutors` would not contain the given `ResourceID`?
---
