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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
