You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2023/07/06 12:42:15 UTC
[flink] 01/04: [hotfix][network] Remove partition file writer and reader for TierFactory#createProducerAgent
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 13d09004de44d5bc91729a1d5671e644f5cbfef3
Author: Yuxin Tan <ta...@gmail.com>
AuthorDate: Mon Jul 3 20:22:51 2023 +0800
[hotfix][network] Remove partition file writer and reader for TierFactory#createProducerAgent
---
.../hybrid/tiered/common/TieredStorageUtils.java | 2 --
.../tiered/file/ProducerMergedPartitionFile.java | 2 ++
.../partition/hybrid/tiered/tier/TierFactory.java | 4 ----
.../hybrid/tiered/tier/disk/DiskTierFactory.java | 23 +++++++++++++++++-----
.../tiered/tier/disk/DiskTierProducerAgent.java | 6 ++----
.../tiered/tier/memory/MemoryTierFactory.java | 4 ----
.../hybrid/tiered/storage/TestingTierFactory.java | 4 ----
.../tier/disk/DiskTierProducerAgentTest.java | 7 ++++---
8 files changed, 26 insertions(+), 26 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageUtils.java
index 7c351a81aec..615cde5da98 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageUtils.java
@@ -28,8 +28,6 @@ import java.util.List;
/** Utils for reading or writing from tiered storage. */
public class TieredStorageUtils {
- public static final String DATA_FILE_SUFFIX = ".tier-storage.data";
-
public static ByteBuffer[] generateBufferWithHeaders(
List<Tuple2<Buffer, Integer>> bufferWithIndexes) {
ByteBuffer[] bufferWithHeaders = new ByteBuffer[2 * bufferWithIndexes.size()];
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFile.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFile.java
index 3561b35dd61..fef384286b7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFile.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFile.java
@@ -26,6 +26,8 @@ import java.nio.file.Path;
*/
public class ProducerMergedPartitionFile {
+ public static final String DATA_FILE_SUFFIX = ".tier-storage.data";
+
public static ProducerMergedPartitionFileWriter createPartitionFileWriter(
Path dataFilePath, ProducerMergedPartitionFileIndex partitionFileIndex) {
return new ProducerMergedPartitionFileWriter(dataFilePath, partitionFileIndex);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
index b5e6d3d0e3e..fabdd57992c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
@@ -20,8 +20,6 @@ package org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier;
import org.apache.flink.runtime.io.disk.BatchShuffleReadBufferPool;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
-import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.PartitionFileReader;
-import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.PartitionFileWriter;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.netty.TieredStorageNettyService;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageConsumerSpec;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageMemoryManager;
@@ -43,8 +41,6 @@ public interface TierFactory {
TieredStoragePartitionId partitionID,
String dataFileBasePath,
boolean isBroadcastOnly,
- PartitionFileWriter partitionFileWriter,
- PartitionFileReader partitionFileReader,
TieredStorageMemoryManager storageMemoryManager,
TieredStorageNettyService nettyService,
TieredStorageResourceRegistry resourceRegistry,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierFactory.java
index f06d6ddd04f..65288fe01a6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierFactory.java
@@ -20,8 +20,10 @@ package org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk;
import org.apache.flink.runtime.io.disk.BatchShuffleReadBufferPool;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
-import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.PartitionFileReader;
-import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.PartitionFileWriter;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.ProducerMergedPartitionFile;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.ProducerMergedPartitionFileIndex;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.ProducerMergedPartitionFileReader;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.ProducerMergedPartitionFileWriter;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.netty.TieredStorageNettyService;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageConsumerSpec;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageMemoryManager;
@@ -32,10 +34,14 @@ import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierFact
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierMasterAgent;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
+import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.ProducerMergedPartitionFile.DATA_FILE_SUFFIX;
+
/** The implementation of {@link TierFactory} for disk tier. */
public class DiskTierFactory implements TierFactory {
@@ -63,8 +69,6 @@ public class DiskTierFactory implements TierFactory {
TieredStoragePartitionId partitionId,
String dataFileBasePath,
boolean isBroadcastOnly,
- PartitionFileWriter partitionFileWriter,
- PartitionFileReader partitionFileReader,
TieredStorageMemoryManager storageMemoryManager,
TieredStorageNettyService nettyService,
TieredStorageResourceRegistry resourceRegistry,
@@ -73,12 +77,21 @@ public class DiskTierFactory implements TierFactory {
int maxRequestedBuffers,
Duration bufferRequestTimeout,
int maxBufferReadAhead) {
+ ProducerMergedPartitionFileIndex partitionFileIndex =
+ new ProducerMergedPartitionFileIndex(isBroadcastOnly ? 1 : numSubpartitions);
+ Path dataFilePath = Paths.get(dataFileBasePath + DATA_FILE_SUFFIX);
+ ProducerMergedPartitionFileWriter partitionFileWriter =
+ ProducerMergedPartitionFile.createPartitionFileWriter(
+ dataFilePath, partitionFileIndex);
+ ProducerMergedPartitionFileReader partitionFileReader =
+ ProducerMergedPartitionFile.createPartitionFileReader(
+ dataFilePath, partitionFileIndex);
return new DiskTierProducerAgent(
partitionId,
numSubpartitions,
numBytesPerSegment,
bufferSizeBytes,
- dataFileBasePath,
+ dataFilePath,
minReservedDiskSpaceFraction,
isBroadcastOnly,
partitionFileWriter,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierProducerAgent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierProducerAgent.java
index 1346c9378dd..9df1cc2755e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierProducerAgent.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierProducerAgent.java
@@ -41,7 +41,6 @@ import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.nio.file.Paths;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
@@ -49,7 +48,6 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
-import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageUtils.DATA_FILE_SUFFIX;
import static org.apache.flink.util.Preconditions.checkArgument;
/** The disk tier implementation of {@link TierProducerAgent}. */
@@ -86,7 +84,7 @@ public class DiskTierProducerAgent implements TierProducerAgent, NettyServicePro
int numSubpartitions,
int numBytesPerSegment,
int bufferSizeBytes,
- String dataFileBasePath,
+ Path dataFilePath,
float minReservedDiskSpaceFraction,
boolean isBroadcastOnly,
PartitionFileWriter partitionFileWriter,
@@ -106,7 +104,7 @@ public class DiskTierProducerAgent implements TierProducerAgent, NettyServicePro
this.partitionId = partitionId;
this.numBuffersPerSegment = numBytesPerSegment / bufferSizeBytes;
this.bufferSizeBytes = bufferSizeBytes;
- this.dataFilePath = Paths.get(dataFileBasePath + DATA_FILE_SUFFIX);
+ this.dataFilePath = dataFilePath;
this.minReservedDiskSpaceFraction = minReservedDiskSpaceFraction;
this.firstBufferIndexInSegment = new ArrayList<>();
this.currentSubpartitionWriteBuffers = new int[numSubpartitions];
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/memory/MemoryTierFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/memory/MemoryTierFactory.java
index 81a3706a49e..a4b10350d9d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/memory/MemoryTierFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/memory/MemoryTierFactory.java
@@ -20,8 +20,6 @@ package org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.memory;
import org.apache.flink.runtime.io.disk.BatchShuffleReadBufferPool;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
-import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.PartitionFileReader;
-import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.PartitionFileWriter;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.netty.TieredStorageNettyService;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageConsumerSpec;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageMemoryManager;
@@ -60,8 +58,6 @@ public class MemoryTierFactory implements TierFactory {
TieredStoragePartitionId partitionID,
String dataFileBasePath,
boolean isBroadcastOnly,
- PartitionFileWriter partitionFileWriter,
- PartitionFileReader partitionFileReader,
TieredStorageMemoryManager memoryManager,
TieredStorageNettyService nettyService,
TieredStorageResourceRegistry resourceRegistry,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TestingTierFactory.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TestingTierFactory.java
index 29652fc1ca3..16ec4c8b24c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TestingTierFactory.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TestingTierFactory.java
@@ -20,8 +20,6 @@ package org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage;
import org.apache.flink.runtime.io.disk.BatchShuffleReadBufferPool;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
-import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.PartitionFileReader;
-import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.PartitionFileWriter;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.netty.TieredStorageNettyService;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierConsumerAgent;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierFactory;
@@ -70,8 +68,6 @@ public class TestingTierFactory implements TierFactory {
TieredStoragePartitionId partitionID,
String dataFileBasePath,
boolean isBroadcastOnly,
- PartitionFileWriter partitionFileWriter,
- PartitionFileReader partitionFileReader,
TieredStorageMemoryManager storageMemoryManager,
TieredStorageNettyService nettyService,
TieredStorageResourceRegistry resourceRegistry,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierProducerAgentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierProducerAgentTest.java
index 32037c7abda..636f8d0095b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierProducerAgentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierProducerAgentTest.java
@@ -26,7 +26,6 @@ import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingTiered
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageIdMappingUtils;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageSubpartitionId;
-import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageUtils;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.PartitionFileWriter;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.TestingPartitionFileReader;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.TestingPartitionFileWriter;
@@ -44,6 +43,7 @@ import java.nio.file.Path;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;
+import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.ProducerMergedPartitionFile.DATA_FILE_SUFFIX;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -66,7 +66,7 @@ public class DiskTierProducerAgentTest {
@Test
void testStartNewSegmentSuccess() throws IOException {
String partitionFile = TempDirUtils.newFile(tempFolder, "test").toString();
- File testFile = new File(partitionFile + TieredStorageUtils.DATA_FILE_SUFFIX);
+ File testFile = new File(partitionFile + DATA_FILE_SUFFIX);
assertThat(testFile.createNewFile()).isTrue();
try (DiskTierProducerAgent diskTierProducerAgent =
createDiskTierProducerAgent(
@@ -197,13 +197,14 @@ public class DiskTierProducerAgentTest {
TestingNettyServiceProducer nettyServiceProducer =
new TestingNettyServiceProducer.Builder().build();
nettyService.registerProducer(PARTITION_ID, nettyServiceProducer);
+ Path dataFilePath = new File(dataFileBasePath + DATA_FILE_SUFFIX).toPath();
return new DiskTierProducerAgent(
PARTITION_ID,
NUM_SUBPARTITIONS,
numBytesPerSegment,
BUFFER_SIZE_BYTES,
- dataFileBasePath,
+ dataFilePath,
minReservedDiskSpaceFraction,
isBroadcastOnly,
partitionFileWriter,