You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2022/09/30 01:24:43 UTC
[flink] 02/02: [FLINK-29460] fix the unstable test HsResultPartitionTest.testRelease
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit df1681c7ef3f5a946f626c86748a9fe46ee687d7
Author: Weijie Guo <re...@163.com>
AuthorDate: Thu Sep 29 11:03:30 2022 +0800
[FLINK-29460] fix the unstable test HsResultPartitionTest.testRelease
This is caused by FLINK-29425, it changes the conditions for spilling, which may cause some buffers to be released during the test.
This closes #20918
---
.../partition/hybrid/HsResultPartitionTest.java | 32 ++++++++++++++++++----
1 file changed, 26 insertions(+), 6 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartitionTest.java
index 826a7b243f2..70378a453d0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartitionTest.java
@@ -236,13 +236,22 @@ class HsResultPartitionTest {
@Test
@Timeout(30)
void testRelease() throws Exception {
+ final int numSubpartitions = 2;
final int numBuffers = 10;
BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
- HsResultPartition partition = createHsResultPartition(2, bufferPool);
+ HsResultPartition partition =
+ createHsResultPartition(
+ numSubpartitions,
+ bufferPool,
+ HybridShuffleConfiguration.builder(
+ numSubpartitions, readBufferPool.getNumBuffersPerRequest())
+ .setFullStrategyNumBuffersTriggerSpillingRatio(0.6f)
+ .setFullStrategyReleaseBufferRatio(0.8f)
+ .build());
- partition.emitRecord(ByteBuffer.allocate(bufferSize * numBuffers), 1);
- assertThat(bufferPool.bestEffortGetNumOfUsedBuffers()).isEqualTo(numBuffers);
+ partition.emitRecord(ByteBuffer.allocate(bufferSize * 5), 1);
+ assertThat(bufferPool.bestEffortGetNumOfUsedBuffers()).isEqualTo(5);
partition.close();
assertThat(bufferPool.isDestroyed()).isTrue();
@@ -381,6 +390,19 @@ class HsResultPartitionTest {
private HsResultPartition createHsResultPartition(int numSubpartitions, BufferPool bufferPool)
throws IOException {
+ return createHsResultPartition(
+ numSubpartitions,
+ bufferPool,
+ HybridShuffleConfiguration.builder(
+ numSubpartitions, readBufferPool.getNumBuffersPerRequest())
+ .build());
+ }
+
+ private HsResultPartition createHsResultPartition(
+ int numSubpartitions,
+ BufferPool bufferPool,
+ HybridShuffleConfiguration hybridShuffleConfiguration)
+ throws IOException {
HsResultPartition hsResultPartition =
new HsResultPartition(
"HsResultPartitionTest",
@@ -394,9 +416,7 @@ class HsResultPartitionTest {
new ResultPartitionManager(),
fileChannelManager.createChannel().getPath(),
bufferSize,
- HybridShuffleConfiguration.builder(
- numSubpartitions, readBufferPool.getNumBuffersPerRequest())
- .build(),
+ hybridShuffleConfiguration,
null,
() -> bufferPool);
taskIOMetricGroup =