[FLINK-22212][coordination] Refactor resource requirement logging

deepfield · Apr 14, 2021 · df808f1 · df808f1
1 parent da22f75
commit df808f1
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 9 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -564,11 +564,6 @@ public CompletableFuture<Acknowledge> declareRequiredResources(
 
  if (null != jobManagerRegistration) {
  if (Objects.equals(jobMasterId, jobManagerRegistration.getJobMasterId())) {
- log.info(
- "Received resource declaration for job {}: {}",
- jobId,
- resourceRequirements.getResourceRequirements());
-
  slotManager.processResourceRequirements(resourceRequirements);
 
  return CompletableFuture.completedFuture(Acknowledge.get());

diff --git a/...ain/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java b/...ain/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java
@@ -248,10 +248,14 @@ public void close() throws Exception {
  @Override
  public void processResourceRequirements(ResourceRequirements resourceRequirements) {
  checkInit();
- LOG.debug(
- "Received resource requirements from job {}: {}",
- resourceRequirements.getJobId(),
- resourceRequirements.getResourceRequirements());
+ if (resourceRequirements.getResourceRequirements().isEmpty()) {
+ LOG.info("Clearing resource requirements of job {}", resourceRequirements.getJobId());
+ } else {
+ LOG.info(
+ "Received resource requirements from job {}: {}",
+ resourceRequirements.getJobId(),
+ resourceRequirements.getResourceRequirements());
+ }
 
  if (resourceRequirements.getResourceRequirements().isEmpty()) {
  jobMasterTargetAddresses.remove(resourceRequirements.getJobId());