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