[hotfix][task] Add task name to BufferDebloater logging

Treydone · Mar 28, 2022 · 01498ef · 01498ef
1 parent 8f51746
commit 01498ef
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 5 deletions.
diff --git a/...n/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java b/...n/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
@@ -146,18 +146,20 @@ public SingleInputGate create(
                         networkBufferSize,
                         new ThroughputCalculator(SystemClock.getInstance()),
                         maybeCreateBufferDebloater(
-                                gateIndex, networkInputGroup.addGroup(gateIndex)));
+                                owningTaskName, gateIndex, networkInputGroup.addGroup(gateIndex)));
 
         InputChannelMetrics metrics =
                 new InputChannelMetrics(networkInputGroup, owner.getParentGroup());
         createInputChannels(owningTaskName, igdd, inputGate, subpartitionIndexRange, metrics);
         return inputGate;
     }
 
-    private BufferDebloater maybeCreateBufferDebloater(int gateIndex, MetricGroup inputGroup) {
+    private BufferDebloater maybeCreateBufferDebloater(
+            String owningTaskName, int gateIndex, MetricGroup inputGroup) {
         if (debloatConfiguration.isEnabled()) {
             final BufferDebloater bufferDebloater =
                     new BufferDebloater(
+                            owningTaskName,
                             gateIndex,
                             debloatConfiguration.getTargetTotalBufferSize().toMillis(),
                             debloatConfiguration.getMaxBufferSize(),

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloater.java b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloater.java
@@ -34,6 +34,7 @@ public class BufferDebloater {
     private static final Logger LOG = LoggerFactory.getLogger(BufferDebloater.class);
     private static final long MILLIS_IN_SECOND = 1000;
 
+    private final String owningTaskName;
     private final int gateIndex;
     private final long targetTotalBufferSize;
     private final int maxBufferSize;
@@ -45,12 +46,14 @@ public class BufferDebloater {
     private int lastBufferSize;
 
     public BufferDebloater(
+            String owningTaskName,
             int gateIndex,
             long targetTotalBufferSize,
             int maxBufferSize,
             int minBufferSize,
             int bufferDebloatThresholdPercentages,
             long numberOfSamples) {
+        this.owningTaskName = owningTaskName;
         this.gateIndex = gateIndex;
         this.targetTotalBufferSize = targetTotalBufferSize;
         this.maxBufferSize = maxBufferSize;
@@ -61,7 +64,8 @@ public BufferDebloater(
         bufferSizeEMA = new BufferSizeEMA(maxBufferSize, minBufferSize, numberOfSamples);
 
         LOG.debug(
-                "Buffer debloater init settings: gateIndex={}, targetTotalBufferSize={}, maxBufferSize={}, minBufferSize={}, bufferDebloatThresholdPercentages={}, numberOfSamples={}",
+                "{}: Buffer debloater init settings: gateIndex={}, targetTotalBufferSize={}, maxBufferSize={}, minBufferSize={}, bufferDebloatThresholdPercentages={}, numberOfSamples={}",
+                owningTaskName,
                 gateIndex,
                 targetTotalBufferSize,
                 maxBufferSize,
@@ -89,7 +93,8 @@ public OptionalInt recalculateBufferSize(long currentThroughput, int buffersInUs
         boolean skipUpdate = skipUpdate(newSize);
 
         LOG.debug(
-                "Buffer size recalculation: gateIndex={}, currentSize={}, newSize={}, instantThroughput={}, desiredBufferSize={}, buffersInUse={}, estimatedTimeToConsumeBuffers={}, announceNewSize={}",
+                "{}: Buffer size recalculation: gateIndex={}, currentSize={}, newSize={}, instantThroughput={}, desiredBufferSize={}, buffersInUse={}, estimatedTimeToConsumeBuffers={}, announceNewSize={}",
+                owningTaskName,
                 gateIndex,
                 lastBufferSize,
                 newSize,

diff --git a/...t/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java b/...t/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
@@ -190,6 +190,7 @@ public SingleInputGate build() {
     private BufferDebloater maybeCreateBufferDebloater(int gateIndex) {
         if (bufferDebloatConfiguration.isEnabled()) {
             return new BufferDebloater(
+                    "Unknown task name in test",
                     gateIndex,
                     bufferDebloatConfiguration.getTargetTotalBufferSize().toMillis(),
                     bufferDebloatConfiguration.getMaxBufferSize(),

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/BufferDebloaterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/BufferDebloaterTest.java
@@ -274,7 +274,13 @@ public BufferDebloater expectBufferSize(int expectedBufferSize) {
 
         private BufferDebloater getBufferDebloater() {
             return new BufferDebloater(
-                    0, debloatTarget, maxBufferSize, minBufferSize, thresholdPercentages, 1);
+                    "Unknown task name in test",
+                    0,
+                    debloatTarget,
+                    maxBufferSize,
+                    minBufferSize,
+                    thresholdPercentages,
+                    1);
         }
     }
 }