You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2021/08/06 07:34:15 UTC
[flink] branch release-1.13 updated (2619535 -> 568c649)
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a change to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 2619535 Update japicmp configuration for 1.13.2
new 491ff74 [FLINK-23409][runtime][test] Add logs for FineGrainedSlotManager#checkResourceRequirements
new 568c649 [FLINK-22891][runtime] Using CompletableFuture to sync the scheduling of checkResourceRequirements
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../slotmanager/FineGrainedSlotManager.java | 34 +++++++++++++---------
1 file changed, 21 insertions(+), 13 deletions(-)
[flink] 01/02: [FLINK-23409][runtime][test] Add logs for
FineGrainedSlotManager#checkResourceRequirements
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 491ff740661bbe27615ef13ad14eb68e72e30721
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Mon Jul 19 10:45:04 2021 +0800
[FLINK-23409][runtime][test] Add logs for FineGrainedSlotManager#checkResourceRequirements
This closes #16530
---
.../runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
index 5a3a372..04b6ad8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
@@ -490,6 +490,7 @@ public class FineGrainedSlotManager implements SlotManager {
*/
private void checkResourceRequirementsWithDelay() {
if (lastResourceRequirementsCheck == null || lastResourceRequirementsCheck.isDone()) {
+ LOG.info("Scheduling the resource requirement check.");
lastResourceRequirementsCheck =
scheduledExecutor.schedule(
() -> mainThreadExecutor.execute(this::checkResourceRequirements),
@@ -502,6 +503,7 @@ public class FineGrainedSlotManager implements SlotManager {
* DO NOT call this method directly. Use {@link #checkResourceRequirementsWithDelay()} instead.
*/
private void checkResourceRequirements() {
+ LOG.info("Matching resource requirements against available resources.");
Map<JobID, Collection<ResourceRequirement>> missingResources =
resourceTracker.getMissingResources();
if (missingResources.isEmpty()) {
@@ -519,9 +521,11 @@ public class FineGrainedSlotManager implements SlotManager {
missingResources, taskManagerTracker);
// Allocate slots according to the result
+ LOG.info("Allocating slots for the resource requirements.");
allocateSlotsAccordingTo(result.getAllocationsOnRegisteredResources());
// Allocate task managers according to the result
+ LOG.info("Allocating task managers for the resource requirements.");
final Set<PendingTaskManagerId> failAllocations =
allocateTaskManagersAccordingTo(result.getPendingTaskManagersToAllocate());
[flink] 02/02: [FLINK-22891][runtime] Using CompletableFuture to
sync the scheduling of checkResourceRequirements
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 568c649cb1e1aab53e5849c2c2ea7630bcc4da75
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Fri Jul 30 10:42:21 2021 +0800
[FLINK-22891][runtime] Using CompletableFuture to sync the scheduling of checkResourceRequirements
The result of ScheduledFuture#isDone can be false negative
This closes #16707
---
.../slotmanager/FineGrainedSlotManager.java | 30 ++++++++++++----------
1 file changed, 17 insertions(+), 13 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
index 04b6ad8..fdd285f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
@@ -109,7 +109,7 @@ public class FineGrainedSlotManager implements SlotManager {
@Nullable private ScheduledFuture<?> taskManagerTimeoutsCheck;
- @Nullable private ScheduledFuture<?> lastResourceRequirementsCheck;
+ @Nullable private CompletableFuture<Void> requirementsCheckFuture;
/** True iff the component has been started. */
private boolean started;
@@ -146,7 +146,7 @@ public class FineGrainedSlotManager implements SlotManager {
resourceActions = null;
mainThreadExecutor = null;
taskManagerTimeoutsCheck = null;
- lastResourceRequirementsCheck = null;
+ requirementsCheckFuture = null;
started = false;
}
@@ -226,11 +226,6 @@ public class FineGrainedSlotManager implements SlotManager {
taskManagerTimeoutsCheck = null;
}
- if (lastResourceRequirementsCheck != null && !lastResourceRequirementsCheck.isDone()) {
- lastResourceRequirementsCheck.cancel(false);
- lastResourceRequirementsCheck = null;
- }
-
slotStatusSyncer.close();
taskManagerTracker.clear();
resourceTracker.clear();
@@ -489,13 +484,19 @@ public class FineGrainedSlotManager implements SlotManager {
* are performed with a slight delay.
*/
private void checkResourceRequirementsWithDelay() {
- if (lastResourceRequirementsCheck == null || lastResourceRequirementsCheck.isDone()) {
+ if (requirementsCheckFuture == null || requirementsCheckFuture.isDone()) {
LOG.info("Scheduling the resource requirement check.");
- lastResourceRequirementsCheck =
- scheduledExecutor.schedule(
- () -> mainThreadExecutor.execute(this::checkResourceRequirements),
- requirementsCheckDelay.toMilliseconds(),
- TimeUnit.MILLISECONDS);
+ requirementsCheckFuture = new CompletableFuture<>();
+ scheduledExecutor.schedule(
+ () ->
+ mainThreadExecutor.execute(
+ () -> {
+ checkResourceRequirements();
+ Preconditions.checkNotNull(requirementsCheckFuture)
+ .complete(null);
+ }),
+ requirementsCheckDelay.toMilliseconds(),
+ TimeUnit.MILLISECONDS);
}
}
@@ -503,6 +504,9 @@ public class FineGrainedSlotManager implements SlotManager {
* DO NOT call this method directly. Use {@link #checkResourceRequirementsWithDelay()} instead.
*/
private void checkResourceRequirements() {
+ if (!started) {
+ return;
+ }
LOG.info("Matching resource requirements against available resources.");
Map<JobID, Collection<ResourceRequirement>> missingResources =
resourceTracker.getMissingResources();