This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new 53bed69377e [FLINK-40335][runtime] Fix Hybrid Shuffle index corruption
caused by shared ByteBuffer
53bed69377e is described below
commit 53bed69377e89dc1a631fcaaa09684664f462de3
Author: Mate Czagany <[email protected]>
AuthorDate: Tue Aug 18 19:50:32 2026 +0200
[FLINK-40335][runtime] Fix Hybrid Shuffle index corruption caused by shared
ByteBuffer
---
.../file/ProducerMergedPartitionFileIndex.java | 5 +-
.../file/ProducerMergedPartitionFileIndexTest.java | 60 ++++++++++++++++++++++
2 files changed, 61 insertions(+), 4 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileIndex.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileIndex.java
index 3046683f475..eb38d302b6b 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileIndex.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileIndex.java
@@ -87,7 +87,7 @@ public class ProducerMergedPartitionFileIndex {
regionGroupSizeInBytes,
numRetainedInMemoryRegionsMax,
FixedSizeRegion.REGION_SIZE,
-
ProducerMergedPartitionFileDataIndexRegionHelper.INSTANCE));
+ new
ProducerMergedPartitionFileDataIndexRegionHelper()));
}
/**
@@ -234,9 +234,6 @@ public class ProducerMergedPartitionFileIndex {
private final ByteBuffer regionBuffer =
allocateAndConfigureBuffer(FixedSizeRegion.REGION_SIZE);
- static final ProducerMergedPartitionFileDataIndexRegionHelper INSTANCE
=
- new ProducerMergedPartitionFileDataIndexRegionHelper();
-
private ProducerMergedPartitionFileDataIndexRegionHelper() {}
@Override
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileIndexTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileIndexTest.java
index a382511b53a..4625e3a9f44 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileIndexTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileIndexTest.java
@@ -20,18 +20,25 @@ package
org.apache.flink.runtime.io.network.partition.hybrid.tiered.file;
import org.apache.flink.api.java.tuple.Tuple2;
import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageSubpartitionId;
+import org.apache.flink.util.ExecutorUtils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.io.TempDir;
import java.nio.file.Path;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
@@ -68,6 +75,59 @@ class ProducerMergedPartitionFileIndexTest {
assertThat(numExpectedRegions).isEqualTo(numGetRegions);
}
+ @Test
+ @Timeout(60)
+ void testConcurrentPartitionFileIndexes() throws Exception {
+ ProducerMergedPartitionFileIndex firstIndex = createIndex(".index-1");
+ ProducerMergedPartitionFileIndex secondIndex = createIndex(".index-2");
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ final int numIterations = 10_000;
+
+ try {
+ Future<Void> firstTask =
+ executor.submit(() -> repeatedlySpillAndRead(firstIndex,
numIterations));
+ Future<Void> secondTask =
+ executor.submit(() -> repeatedlySpillAndRead(secondIndex,
numIterations));
+
+ firstTask.get();
+ secondTask.get();
+ } finally {
+ ExecutorUtils.gracefulShutdown(10_000L, TimeUnit.MILLISECONDS,
executor);
+ firstIndex.release();
+ secondIndex.release();
+ }
+ }
+
+ private ProducerMergedPartitionFileIndex createIndex(String fileName) {
+ return new ProducerMergedPartitionFileIndex(
+ 1, indexFilePath.resolveSibling(fileName), 256, 1);
+ }
+
+ private static Void repeatedlySpillAndRead(
+ ProducerMergedPartitionFileIndex index, int numIterations) {
+ for (int iteration = 0; iteration < numIterations; iteration++) {
+ int bufferIndex = iteration % 2 * 2;
+ index.addBuffers(
+ Collections.singletonList(
+ new ProducerMergedPartitionFileIndex.FlushedBuffer(
+ 0, bufferIndex, iteration, 1)));
+
+ if (iteration > 0) {
+ // The region written in the previous iteration was just
evicted to the index file
+ // by the addBuffers above (the cache holds a single region),
so this read must go
+ // through the spill file.
+ int previousBufferIndex = (iteration - 1) % 2 * 2;
+ assertThat(index.getRegion(new TieredStorageSubpartitionId(0),
previousBufferIndex))
+ .get()
+ .extracting(
+
ProducerMergedPartitionFileIndex.FixedSizeRegion
+ ::getRegionStartOffset)
+ .isEqualTo((long) iteration - 1);
+ }
+ }
+ return null;
+ }
+
private Tuple2<Integer, Integer> generateFlushedBuffers(
int numSubpartitions,
int numBuffersPerSubpartition,
