Github user Clarkkkkk commented on a diff in the pull request:
https://github.com/apache/flink/pull/6192#discussion_r199190175
--- 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 --
I will work on the implementation of the idea you mentioned above which
makes perfect sense to me.
---
