[GitHub] [flink] dawidwys commented on a change in pull request #17437: [FLINK-24467][streaming] Announce the min and max buffer size despite last diff less than threshold

[GitBox <gi...@apache.org>]

dawidwys commented on a change in pull request #17437:
URL: https://github.com/apache/flink/pull/17437#discussion_r725959369



##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/bufferdebloat/BufferDebloaterTest.java
##########
@@ -105,14 +107,50 @@ public void testNegativeConsumptionTime() {
         testBufferSizeCalculation(3, asList(3, 5, 8), 3333, 50, 1100, -1, 248);
     }
 
-    private void testBufferSizeCalculation(
+    @Test
+    public void testAnnouncedMaxBufferSizeEvenDespiteLastDiffLessThanThreshold() {
+        // Calculate the buffer size a little lower than the max buffer size.
+        BufferDebloater bufferDebloater =
+                testBufferSizeCalculation(1, singletonList(1), 500, 50, 1100, 1000, 500);

Review comment:
       nit: it's rather hard to understand what the numbers mean. Would be nice to have some kind of an asserter/builder pattern here IMO:
   ```
   testBufferDebloat()
      .withGates(1)
      .withNumberOfBuffersInUse(1, ...)
      .withThroughput(...)
      .withBufferSize(min, max)
      .expectBufferSize(500);
   ```

##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/bufferdebloat/BufferDebloaterTest.java
##########
@@ -105,14 +107,50 @@ public void testNegativeConsumptionTime() {
         testBufferSizeCalculation(3, asList(3, 5, 8), 3333, 50, 1100, -1, 248);
     }
 
-    private void testBufferSizeCalculation(
+    @Test
+    public void testAnnouncedMaxBufferSizeEvenDespiteLastDiffLessThanThreshold() {

Review comment:
       ```suggestion
       public void testAnnouncedMaxBufferSizeDespiteLastDiffLessThanThreshold() {
   ```

##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/bufferdebloat/BufferDebloaterTest.java
##########
@@ -105,14 +107,50 @@ public void testNegativeConsumptionTime() {
         testBufferSizeCalculation(3, asList(3, 5, 8), 3333, 50, 1100, -1, 248);
     }
 
-    private void testBufferSizeCalculation(
+    @Test
+    public void testAnnouncedMaxBufferSizeEvenDespiteLastDiffLessThanThreshold() {
+        // Calculate the buffer size a little lower than the max buffer size.
+        BufferDebloater bufferDebloater =
+                testBufferSizeCalculation(1, singletonList(1), 500, 50, 1100, 1000, 500);
+        bufferDebloater.recalculateBufferSize(1000);
+        assertThat(bufferDebloater.getLastBufferSize(), is(1000));
+
+        // Recalculate the buffer size to max value.
+        bufferDebloater.recalculateBufferSize(2000);
+
+        // The max value should be announced despite it differ from the previous one by less than
+        // threshold value.
+        assertThat(bufferDebloater.getLastBufferSize(), is(1100));
+
+        // Make sure that there is no repeated announcement of max buffer size.
+        bufferDebloater.recalculateBufferSize(2000);
+    }
+
+    @Test
+    public void testAnnouncedMinBufferSizeEvenDespiteLastDiffLessThanThreshold() {
+        // Calculate the buffer size a little greater than the min buffer size.
+        BufferDebloater bufferDebloater =
+                testBufferSizeCalculation(1, singletonList(1), 60, 50, 1100, 1000, 60);

Review comment:
       What is the initial buffer size? Can we make it prominent in the test similarly as in the max buffer size case.

##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/bufferdebloat/BufferDebloaterTest.java
##########
@@ -105,14 +107,50 @@ public void testNegativeConsumptionTime() {
         testBufferSizeCalculation(3, asList(3, 5, 8), 3333, 50, 1100, -1, 248);
     }
 
-    private void testBufferSizeCalculation(
+    @Test
+    public void testAnnouncedMaxBufferSizeEvenDespiteLastDiffLessThanThreshold() {
+        // Calculate the buffer size a little lower than the max buffer size.
+        BufferDebloater bufferDebloater =
+                testBufferSizeCalculation(1, singletonList(1), 500, 50, 1100, 1000, 500);
+        bufferDebloater.recalculateBufferSize(1000);
+        assertThat(bufferDebloater.getLastBufferSize(), is(1000));
+
+        // Recalculate the buffer size to max value.
+        bufferDebloater.recalculateBufferSize(2000);
+
+        // The max value should be announced despite it differ from the previous one by less than
+        // threshold value.
+        assertThat(bufferDebloater.getLastBufferSize(), is(1100));
+
+        // Make sure that there is no repeated announcement of max buffer size.
+        bufferDebloater.recalculateBufferSize(2000);
+    }
+
+    @Test
+    public void testAnnouncedMinBufferSizeEvenDespiteLastDiffLessThanThreshold() {

Review comment:
       ```suggestion
       public void testAnnouncedMinBufferSizeDespiteLastDiffLessThanThreshold() {
   ```

##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/bufferdebloat/BufferDebloaterTest.java
##########
@@ -105,14 +107,50 @@ public void testNegativeConsumptionTime() {
         testBufferSizeCalculation(3, asList(3, 5, 8), 3333, 50, 1100, -1, 248);
     }
 
-    private void testBufferSizeCalculation(
+    @Test
+    public void testAnnouncedMaxBufferSizeEvenDespiteLastDiffLessThanThreshold() {
+        // Calculate the buffer size a little lower than the max buffer size.
+        BufferDebloater bufferDebloater =
+                testBufferSizeCalculation(1, singletonList(1), 500, 50, 1100, 1000, 500);

Review comment:
       I don't mind if we don't do it in this PR. I put it here more for the future.




-- 
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-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org