[flink] branch master updated: [hotfix] Avoid excessive logging in DeclarativeSlotPoolBridge

[ro...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 212dacd  [hotfix] Avoid excessive logging in DeclarativeSlotPoolBridge
212dacd is described below

commit 212dacdcab69fa030435e60345948d17d60237af
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Jul 14 09:56:45 2021 +0200

    [hotfix] Avoid excessive logging in DeclarativeSlotPoolBridge
---
 .../slotpool/DeclarativeSlotPoolBridge.java          | 20 ++++++++++----------
 1 file changed, 10 insertions(+), 10 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java
index 1d281bb..41d0940 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java
@@ -345,17 +345,17 @@ public class DeclarativeSlotPoolBridge extends DeclarativeSlotPoolService implem
     }
 
     private void failPendingRequests(Collection<ResourceRequirement> acquiredResources) {
-        if (!pendingRequests.isEmpty()) {
-            final NoResourceAvailableException cause =
-                    new NoResourceAvailableException(
-                            "Could not acquire the minimum required resources. Acquired: "
-                                    + acquiredResources
-                                    + ". Current slot pool status: "
-                                    + getSlotServiceStatus());
-
+        Predicate<PendingRequest> predicate =
+                request -> !isBatchSlotRequestTimeoutCheckDisabled || !request.isBatchRequest();
+        if (pendingRequests.values().stream().anyMatch(predicate)) {
+            log.warn(
+                    "Could not acquire the minimum required resources, failing slot requests. Acquired: {}. Current slot pool status: {}",
+                    acquiredResources,
+                    getSlotServiceStatus());
             cancelPendingRequests(
-                    request -> !isBatchSlotRequestTimeoutCheckDisabled || !request.isBatchRequest(),
-                    cause);
+                    predicate,
+                    new NoResourceAvailableException(
+                            "Could not acquire the minimum required resources."));
         }
     }