KarmaGYZ commented on code in PR #23635:
URL: https://github.com/apache/flink/pull/23635#discussion_r1432179061
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:
##########
@@ -334,7 +334,9 @@ public void onUnknownDeploymentsOf(
.createSlotPoolService(
jid,
createDeclarativeSlotPoolFactory(
-
jobMasterConfiguration.getConfiguration()));
+
jobMasterConfiguration.getConfiguration()),
+ null,
Review Comment:
Why not set it according to the configuration?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclarativeSlotPool.java:
##########
@@ -118,26 +129,47 @@ public DefaultDeclarativeSlotPool(
this.totalResourceRequirements = ResourceCounter.empty();
this.fulfilledResourceRequirements = ResourceCounter.empty();
this.slotToRequirementProfileMappings = new HashMap<>();
+ this.componentMainThreadExecutor =
Preconditions.checkNotNull(componentMainThreadExecutor);
+ this.slotRequestMaxInterval = slotRequestMaxInterval;
}
@Override
public void increaseResourceRequirementsBy(ResourceCounter increment) {
- if (increment.isEmpty()) {
+ updateResourceRequirementsBy(
+ increment,
+ () -> totalResourceRequirements =
totalResourceRequirements.add(increment));
+ }
+
+ private void updateResourceRequirementsBy(
+ @Nonnull ResourceCounter deltaResourceCount, @Nonnull Runnable
runnable) {
+ if (deltaResourceCount.isEmpty()) {
return;
}
- totalResourceRequirements = totalResourceRequirements.add(increment);
- declareResourceRequirements();
- }
+ runnable.run();
- @Override
- public void decreaseResourceRequirementsBy(ResourceCounter decrement) {
- if (decrement.isEmpty()) {
+ if (slotRequestMaxInterval == null) {
+ declareResourceRequirements();
return;
}
- totalResourceRequirements =
totalResourceRequirements.subtract(decrement);
- declareResourceRequirements();
+ if (slotRequestMaxIntervalTimeoutFuture != null
+ && !slotRequestMaxIntervalTimeoutFuture.isDone()
+ && !slotRequestMaxIntervalTimeoutFuture.isCancelled()) {
+ slotRequestMaxIntervalTimeoutFuture.cancel(true);
+ }
+ slotRequestMaxIntervalTimeoutFuture =
+ componentMainThreadExecutor.schedule(
+ this::declareResourceRequirements,
+ slotRequestMaxInterval.toMilliseconds(),
+ TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public void decreaseResourceRequirementsBy(ResourceCounter decrement) {
+ updateResourceRequirementsBy(
+ decrement,
+ () -> totalResourceRequirements =
totalResourceRequirements.subtract(decrement));
Review Comment:
ditto
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:
##########
@@ -336,7 +336,8 @@ public void onUnknownDeploymentsOf(
createDeclarativeSlotPoolFactory(
jobMasterConfiguration.getConfiguration()),
null,
- getMainThreadExecutor());
+ getMainThreadExecutor(),
+ false);
Review Comment:
Why not set slotBatchAllocatable according to the configuration?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclarativeSlotPool.java:
##########
@@ -118,26 +129,47 @@ public DefaultDeclarativeSlotPool(
this.totalResourceRequirements = ResourceCounter.empty();
this.fulfilledResourceRequirements = ResourceCounter.empty();
this.slotToRequirementProfileMappings = new HashMap<>();
+ this.componentMainThreadExecutor =
Preconditions.checkNotNull(componentMainThreadExecutor);
+ this.slotRequestMaxInterval = slotRequestMaxInterval;
}
@Override
public void increaseResourceRequirementsBy(ResourceCounter increment) {
- if (increment.isEmpty()) {
+ updateResourceRequirementsBy(
+ increment,
+ () -> totalResourceRequirements =
totalResourceRequirements.add(increment));
Review Comment:
Why not check whether the ResourceCounter is empty and modify the
totalResourceRequirements right here?
--
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.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
