Github user Clarkkkkk commented on a diff in the pull request:
https://github.com/apache/flink/pull/6192#discussion_r199189703
--- 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 --
Yes, you are right. I have been thinking about how to add a global version
state to a container all the week, but feels like I came into a dead end. I
found this question when there are higher priority batch job in yarn cluster
which occupied all the resources and made yarn kill the flink task manager
containers frequently in a period of time.
---
