You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/06/12 17:07:16 UTC
[GitHub] [flink] tillrohrmann commented on a change in pull request #12620: [FLINK-18226][runtime] Fix ActiveResourceManager requesting extra workers on termination of existing workers.
tillrohrmann commented on a change in pull request #12620:
URL: https://github.com/apache/flink/pull/12620#discussion_r439463764
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
##########
@@ -104,13 +104,15 @@
/**
* Sends the given {@link SlotReport} to the ResourceManager.
+ * This method is only used for sending the initial slot report after a TaskManager registered to the ResourceManager.
+ * After that, the heartbeat payload is used for periodically updating the slot reports.
*
* @param taskManagerRegistrationId id identifying the sending TaskManager
* @param slotReport which is sent to the ResourceManager
* @param timeout for the operation
* @return Future which is completed with {@link Acknowledge} once the slot report has been received.
*/
- CompletableFuture<Acknowledge> sendSlotReport(
+ CompletableFuture<Acknowledge> sendInitialSlotReport(
Review comment:
I am a bit unsure about this change since we are naming this method now how it is used by the `TaskExecutor`. However, it speaks nothing against calling `sendSlotReport` multiple times after the `TaskExecutor` has registered at the `RM`. It's just that the `TaskExecutor` has been implemented in the way that it calls this method only once. Hence, I would suggest to revert this change.
##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesResourceManagerTest.java
##########
@@ -272,6 +273,43 @@ public void testTaskManagerPodTerminated() throws Exception {
}};
}
+ @Test
+ public void testTaskManagerPodTerminatedAfterRegistered() throws Exception {
Review comment:
```suggestion
public void testTaskManagerPodTerminatedAfterRegistration() throws Exception {
```
##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesResourceManagerTest.java
##########
@@ -390,31 +429,7 @@ public void testPreviousAttemptPodAdded() throws Exception {
// adding current attempt pod should decrease the pending worker count
resourceManager.onAdded(Collections.singletonList(new KubernetesPod(currentAttemptPod)));
- assertThat(resourceManager.getNumPendingWorkersForTesting(), is(0));
- });
- }};
- }
-
- @Test
- public void testDuplicatedPodAdded() throws Exception {
Review comment:
Why can we remove this test?
##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesResourceManagerTest.java
##########
@@ -272,6 +273,43 @@ public void testTaskManagerPodTerminated() throws Exception {
}};
}
+ @Test
+ public void testTaskManagerPodTerminatedAfterRegistered() throws Exception {
+ new Context() {{
+ runTest(() -> {
+ registerSlotRequest();
+ final Pod pod1 = kubeClient.pods().list().getItems().get(0);
+ resourceManager.onAdded(Collections.singletonList(new KubernetesPod(pod1)));
+ registerTaskExecutor(new ResourceID(pod1.getMetadata().getName()));
+
+ // Terminate the pod. Should not request a new pod.
+ terminatePod(pod1);
+ resourceManager.onModified(Collections.singletonList(new KubernetesPod(pod1)));
+ assertEquals(0, kubeClient.pods().list().getItems().size());
+
+ registerSlotRequest();
+ final Pod pod2 = kubeClient.pods().list().getItems().get(0);
+ resourceManager.onAdded(Collections.singletonList(new KubernetesPod(pod2)));
+ registerTaskExecutor(new ResourceID(pod2.getMetadata().getName()));
+
+ // Error happens in the pod. Should not request a new pod.
+ terminatePod(pod2);
+ resourceManager.onError(Collections.singletonList(new KubernetesPod(pod2)));
+ assertEquals(0, kubeClient.pods().list().getItems().size());
+
+ registerSlotRequest();
+ final Pod pod3 = kubeClient.pods().list().getItems().get(0);
+ resourceManager.onAdded(Collections.singletonList(new KubernetesPod(pod3)));
+ registerTaskExecutor(new ResourceID(pod3.getMetadata().getName()));
+
+ // Delete the pod. Should not request a new pod.
+ terminatePod(pod3);
+ resourceManager.onDeleted(Collections.singletonList(new KubernetesPod(pod3)));
+ assertEquals(0, kubeClient.pods().list().getItems().size());
Review comment:
I think it would be better to split this test case into three separate ones.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ActiveResourceManager.java
##########
@@ -110,39 +118,118 @@ public ActiveResourceManager(
protected abstract Configuration loadClientConfiguration();
- protected int getNumPendingWorkers() {
- return pendingWorkerCounter.getTotalNum();
+ @Override
+ protected void registerTaskManagerAtSlotManager(
+ WorkerRegistration<WorkerType> workerTypeWorkerRegistration, SlotReport slotReport) {
+ super.registerTaskManagerAtSlotManager(workerTypeWorkerRegistration, slotReport);
+ notifyAllocatedWorkerRegistered(workerTypeWorkerRegistration.getResourceID());
+ }
+
+ protected int getNumRequestedNotAllocatedWorkers() {
+ return requestedNotAllocatedWorkerCounter.getTotalNum();
+ }
+
+ protected int getNumRequestedNotAllocatedWorkersFor(WorkerResourceSpec workerResourceSpec) {
+ return requestedNotAllocatedWorkerCounter.getNum(workerResourceSpec);
+ }
+
+ protected int getNumRequestedNotRegisteredWorkers() {
+ return requestedNotRegisteredWorkerCounter.getTotalNum();
}
- protected int getNumPendingWorkersFor(WorkerResourceSpec workerResourceSpec) {
- return pendingWorkerCounter.getNum(workerResourceSpec);
+ protected int getNumRequestedNotRegisteredWorkersFor(WorkerResourceSpec workerResourceSpec) {
+ return requestedNotRegisteredWorkerCounter.getNum(workerResourceSpec);
}
/**
* Notify that a worker with the given resource spec has been requested.
* @param workerResourceSpec resource spec of the requested worker
* @return updated number of pending workers for the given resource spec
*/
- protected int notifyNewWorkerRequested(WorkerResourceSpec workerResourceSpec) {
- return pendingWorkerCounter.increaseAndGet(workerResourceSpec);
+ protected PendingWorkerNums notifyNewWorkerRequested(WorkerResourceSpec workerResourceSpec) {
+ return new PendingWorkerNums(
+ requestedNotAllocatedWorkerCounter.increaseAndGet(workerResourceSpec),
+ requestedNotRegisteredWorkerCounter.increaseAndGet(workerResourceSpec));
}
/**
* Notify that a worker with the given resource spec has been allocated.
* @param workerResourceSpec resource spec of the requested worker
+ * @param resourceID id of the allocated resource
* @return updated number of pending workers for the given resource spec
*/
- protected int notifyNewWorkerAllocated(WorkerResourceSpec workerResourceSpec) {
- return pendingWorkerCounter.decreaseAndGet(workerResourceSpec);
+ protected PendingWorkerNums notifyNewWorkerAllocated(WorkerResourceSpec workerResourceSpec, ResourceID resourceID) {
+ allocatedNotRegisteredWorkerResourceSpecs.put(resourceID, workerResourceSpec);
+ return new PendingWorkerNums(
+ requestedNotAllocatedWorkerCounter.decreaseAndGet(workerResourceSpec),
+ requestedNotRegisteredWorkerCounter.getNum(workerResourceSpec));
}
/**
* Notify that allocation of a worker with the given resource spec has failed.
* @param workerResourceSpec resource spec of the requested worker
* @return updated number of pending workers for the given resource spec
*/
- protected int notifyNewWorkerAllocationFailed(WorkerResourceSpec workerResourceSpec) {
- return pendingWorkerCounter.decreaseAndGet(workerResourceSpec);
+ protected PendingWorkerNums notifyNewWorkerAllocationFailed(WorkerResourceSpec workerResourceSpec) {
+ return new PendingWorkerNums(
+ requestedNotAllocatedWorkerCounter.decreaseAndGet(workerResourceSpec),
+ requestedNotRegisteredWorkerCounter.decreaseAndGet(workerResourceSpec));
+ }
+
+ /**
+ * Notify that a worker with the given resource spec has been registered.
+ * @param resourceID id of the registered worker resource
+ * @return updated number of pending workers for the corresponding resource spec
+ */
+ private PendingWorkerNums notifyAllocatedWorkerRegistered(ResourceID resourceID) {
Review comment:
I think this methods does not need to return something. Then we could get rid of `PendingWorkerNums.ZERO` which does not seem to be a correct return value.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
##########
@@ -404,13 +404,17 @@ private void stopResourceManagerServices() throws Exception {
final WorkerRegistration<WorkerType> workerTypeWorkerRegistration = taskExecutors.get(taskManagerResourceId);
if (workerTypeWorkerRegistration.getInstanceID().equals(taskManagerRegistrationId)) {
- slotManager.registerTaskManager(workerTypeWorkerRegistration, slotReport);
+ registerTaskManagerAtSlotManager(workerTypeWorkerRegistration, slotReport);
Review comment:
I think it would be slightly nicer to do it the following way:
```suggestion
if (slotManager.registerTaskManager(workerTypeWorkerRegistration, slotReport)) {
onTaskManagerRegistration(workerTypeWorkerRegistration);
}
```
That way, sub classes won't be able to obstruct the registration of TMs at the `SlotManager` but can still listen to registration events. Of course, this change would required that `SlotManager.registerTaskManager` returns `true` if it is a new registration and `false` if it is not a new registration.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ActiveResourceManager.java
##########
@@ -110,39 +118,118 @@ public ActiveResourceManager(
protected abstract Configuration loadClientConfiguration();
- protected int getNumPendingWorkers() {
- return pendingWorkerCounter.getTotalNum();
+ @Override
+ protected void registerTaskManagerAtSlotManager(
+ WorkerRegistration<WorkerType> workerTypeWorkerRegistration, SlotReport slotReport) {
+ super.registerTaskManagerAtSlotManager(workerTypeWorkerRegistration, slotReport);
+ notifyAllocatedWorkerRegistered(workerTypeWorkerRegistration.getResourceID());
+ }
+
+ protected int getNumRequestedNotAllocatedWorkers() {
+ return requestedNotAllocatedWorkerCounter.getTotalNum();
+ }
+
+ protected int getNumRequestedNotAllocatedWorkersFor(WorkerResourceSpec workerResourceSpec) {
+ return requestedNotAllocatedWorkerCounter.getNum(workerResourceSpec);
+ }
+
+ protected int getNumRequestedNotRegisteredWorkers() {
+ return requestedNotRegisteredWorkerCounter.getTotalNum();
}
- protected int getNumPendingWorkersFor(WorkerResourceSpec workerResourceSpec) {
- return pendingWorkerCounter.getNum(workerResourceSpec);
+ protected int getNumRequestedNotRegisteredWorkersFor(WorkerResourceSpec workerResourceSpec) {
+ return requestedNotRegisteredWorkerCounter.getNum(workerResourceSpec);
}
/**
* Notify that a worker with the given resource spec has been requested.
* @param workerResourceSpec resource spec of the requested worker
* @return updated number of pending workers for the given resource spec
*/
- protected int notifyNewWorkerRequested(WorkerResourceSpec workerResourceSpec) {
- return pendingWorkerCounter.increaseAndGet(workerResourceSpec);
+ protected PendingWorkerNums notifyNewWorkerRequested(WorkerResourceSpec workerResourceSpec) {
+ return new PendingWorkerNums(
+ requestedNotAllocatedWorkerCounter.increaseAndGet(workerResourceSpec),
+ requestedNotRegisteredWorkerCounter.increaseAndGet(workerResourceSpec));
}
/**
* Notify that a worker with the given resource spec has been allocated.
* @param workerResourceSpec resource spec of the requested worker
+ * @param resourceID id of the allocated resource
* @return updated number of pending workers for the given resource spec
*/
- protected int notifyNewWorkerAllocated(WorkerResourceSpec workerResourceSpec) {
- return pendingWorkerCounter.decreaseAndGet(workerResourceSpec);
+ protected PendingWorkerNums notifyNewWorkerAllocated(WorkerResourceSpec workerResourceSpec, ResourceID resourceID) {
+ allocatedNotRegisteredWorkerResourceSpecs.put(resourceID, workerResourceSpec);
+ return new PendingWorkerNums(
+ requestedNotAllocatedWorkerCounter.decreaseAndGet(workerResourceSpec),
+ requestedNotRegisteredWorkerCounter.getNum(workerResourceSpec));
}
/**
* Notify that allocation of a worker with the given resource spec has failed.
* @param workerResourceSpec resource spec of the requested worker
* @return updated number of pending workers for the given resource spec
*/
- protected int notifyNewWorkerAllocationFailed(WorkerResourceSpec workerResourceSpec) {
- return pendingWorkerCounter.decreaseAndGet(workerResourceSpec);
+ protected PendingWorkerNums notifyNewWorkerAllocationFailed(WorkerResourceSpec workerResourceSpec) {
+ return new PendingWorkerNums(
+ requestedNotAllocatedWorkerCounter.decreaseAndGet(workerResourceSpec),
+ requestedNotRegisteredWorkerCounter.decreaseAndGet(workerResourceSpec));
+ }
+
+ /**
+ * Notify that a worker with the given resource spec has been registered.
+ * @param resourceID id of the registered worker resource
+ * @return updated number of pending workers for the corresponding resource spec
+ */
+ private PendingWorkerNums notifyAllocatedWorkerRegistered(ResourceID resourceID) {
+ WorkerResourceSpec workerResourceSpec = allocatedNotRegisteredWorkerResourceSpecs.remove(resourceID);
+ if (workerResourceSpec == null) {
+ // ignore workers from previous attempt
+ return PendingWorkerNums.ZERO;
+ }
+ return new PendingWorkerNums(
+ requestedNotAllocatedWorkerCounter.getNum(workerResourceSpec),
+ requestedNotRegisteredWorkerCounter.decreaseAndGet(workerResourceSpec));
+ }
+
+ /**
+ * Notify that a worker with the given resource spec has been stopped.
+ * @param resourceID id of the stopped worker resource
+ * @return updated number of pending workers for the corresponding resource spec
+ */
+ protected PendingWorkerNums notifyAllocatedWorkerStopped(ResourceID resourceID) {
Review comment:
Same here. This method does not need to return `PendingWorkerNums`.
----------------------------------------------------------------
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:
users@infra.apache.org