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/08/07 06:26:41 UTC
[flink] branch master updated: [FLINK-32708][network] Fix the write logic in remote tier of hybrid shuffle
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
The following commit(s) were added to refs/heads/master by this push:
new 6e138f1a483 [FLINK-32708][network] Fix the write logic in remote tier of hybrid shuffle
6e138f1a483 is described below
commit 6e138f1a4832bd4379fa6289f0445f1ede9b5a54
Author: Wencong Liu <li...@163.com>
AuthorDate: Fri Jul 28 16:31:17 2023 +0800
[FLINK-32708][network] Fix the write logic in remote tier of hybrid shuffle
---
.../tiered/file/SegmentPartitionFileWriter.java | 4 +-
.../tiered/tier/remote/RemoteStorageScanner.java | 59 +++++++++++++++++-----
.../tier/remote/RemoteStorageScannerTest.java | 5 --
3 files changed, 49 insertions(+), 19 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/SegmentPartitionFileWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/SegmentPartitionFileWriter.java
index 003293741e2..24fe59bfaef 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/SegmentPartitionFileWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/SegmentPartitionFileWriter.java
@@ -178,13 +178,13 @@ public class SegmentPartitionFileWriter implements PartitionFileWriter {
private void writeSegmentFinishFile(
TieredStoragePartitionId partitionId, int subpartitionId, int segmentId) {
try {
- SegmentPartitionFile.writeSegmentFinishFile(
- basePath, partitionId, subpartitionId, segmentId);
WritableByteChannel channel = subpartitionChannels[subpartitionId];
if (channel != null) {
channel.close();
subpartitionChannels[subpartitionId] = null;
}
+ SegmentPartitionFile.writeSegmentFinishFile(
+ basePath, partitionId, subpartitionId, segmentId);
} catch (IOException exception) {
ExceptionUtils.rethrow(exception);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScanner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScanner.java
index 94111825f15..00358b34fe9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScanner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScanner.java
@@ -29,6 +29,9 @@ import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FatalExitExceptionHandler;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import java.io.IOException;
@@ -38,12 +41,12 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.SegmentPartitionFile.getSegmentFinishDirPath;
import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.SegmentPartitionFile.getSegmentPath;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
-import static org.apache.flink.util.Preconditions.checkState;
/**
* The {@link RemoteStorageScanner} is introduced to notify asynchronously for file reading on
@@ -56,6 +59,11 @@ import static org.apache.flink.util.Preconditions.checkState;
*/
public class RemoteStorageScanner implements Runnable {
+ private static final Logger LOG = LoggerFactory.getLogger(RemoteStorageScanner.class);
+
+ /** The max retry time of checking file status through remote filesystem. */
+ private static final int MAX_RETRY_TIME = 100;
+
/** The initial scan interval is 100ms. */
private static final int INITIAL_SCAN_INTERVAL_MS = 100;
@@ -88,6 +96,8 @@ public class RemoteStorageScanner implements Runnable {
private int lastInterval = INITIAL_SCAN_INTERVAL_MS;
+ private int currentRetryTime = 0;
+
public RemoteStorageScanner(String baseRemoteStoragePath) {
this.baseRemoteStoragePath = baseRemoteStoragePath;
this.requiredSegmentIds = new ConcurrentHashMap<>();
@@ -109,7 +119,11 @@ public class RemoteStorageScanner implements Runnable {
/** Start the executor. */
public void start() {
- scannerExecutor.schedule(this, lastInterval, TimeUnit.MILLISECONDS);
+ synchronized (scannerExecutor) {
+ if (!scannerExecutor.isShutdown()) {
+ scannerExecutor.schedule(this, lastInterval, TimeUnit.MILLISECONDS);
+ }
+ }
}
/**
@@ -144,7 +158,16 @@ public class RemoteStorageScanner implements Runnable {
/** Close the executor. */
public void close() {
- scannerExecutor.shutdownNow();
+ synchronized (scannerExecutor) {
+ scannerExecutor.shutdownNow();
+ }
+ try {
+ if (!scannerExecutor.awaitTermination(5L, TimeUnit.MINUTES)) {
+ throw new TimeoutException("Timeout to shutdown the flush thread.");
+ }
+ } catch (InterruptedException | TimeoutException e) {
+ ExceptionUtils.rethrow(e);
+ }
}
/** Iterate the watched segment ids and check related file status. */
@@ -207,20 +230,23 @@ public class RemoteStorageScanner implements Runnable {
Path segmentFinishDir =
getSegmentFinishDirPath(
baseRemoteStoragePath, partitionId, subpartitionId.getSubpartitionId());
- FileStatus[] fileStatuses;
+ FileStatus[] fileStatuses = new FileStatus[0];
try {
if (!remoteFileSystem.exists(segmentFinishDir)) {
return;
}
fileStatuses = remoteFileSystem.listStatus(segmentFinishDir);
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to list the segment finish file. " + segmentFinishDir, e);
+ currentRetryTime = 0;
+ } catch (Throwable t) {
+ if (t instanceof java.io.FileNotFoundException) {
+ return;
+ }
+ currentRetryTime++;
+ tryThrowException(t, "Failed to list the segment finish file.");
}
- if (fileStatuses.length == 0) {
+ if (fileStatuses.length != 1) {
return;
}
- checkState(fileStatuses.length == 1);
scannedMaxSegmentIds.put(
Tuple2.of(partitionId, subpartitionId),
Integer.parseInt(fileStatuses[0].getPath().getName()));
@@ -236,15 +262,24 @@ public class RemoteStorageScanner implements Runnable {
partitionId,
subpartitionId.getSubpartitionId(),
segmentId);
- boolean isExist;
+ boolean isExist = false;
try {
isExist = remoteFileSystem.exists(segmentPath);
- } catch (IOException e) {
- throw new RuntimeException("Failed to check segment file. " + segmentPath, e);
+ currentRetryTime = 0;
+ } catch (Throwable t) {
+ currentRetryTime++;
+ tryThrowException(t, "Failed to check the status of segment file.");
}
return isExist;
}
+ private void tryThrowException(Throwable t, String logMessage) {
+ LOG.warn(logMessage);
+ if (currentRetryTime > MAX_RETRY_TIME) {
+ throw new RuntimeException(logMessage, t);
+ }
+ }
+
/**
* The strategy is used to decide the scan interval of {@link RemoteStorageScanner}. The
* interval will be updated at a double rate and restricted by max value.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScannerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScannerTest.java
index b813ab7fc8f..9b207b04da2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScannerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScannerTest.java
@@ -34,12 +34,10 @@ import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.RejectedExecutionException;
import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.SegmentPartitionFile.getSegmentPath;
import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.SegmentPartitionFile.writeSegmentFinishFile;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
/** Tests for {@link RemoteStorageScanner}. */
class RemoteStorageScannerTest {
@@ -123,9 +121,6 @@ class RemoteStorageScannerTest {
remoteStorageScanner.watchSegment(DEFAULT_PARTITION_ID, DEFAULT_SUBPARTITION_ID, 0);
remoteStorageScanner.start();
future.get();
- remoteStorageScanner.close();
- assertThatThrownBy(remoteStorageScanner::start)
- .isInstanceOf(RejectedExecutionException.class);
}
@Test