You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2025/07/07 13:49:06 UTC
(flink) branch master updated: [FLINK-38056][s3] Improve interruption handling for s5cmd
This is an automated email from the ASF dual-hosted git repository.
pnowojski 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 fa416342d5c [FLINK-38056][s3] Improve interruption handling for s5cmd
fa416342d5c is described below
commit fa416342d5cc44b6b4e1e7a0a62a6634aa152705
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon May 19 14:47:33 2025 +0200
[FLINK-38056][s3] Improve interruption handling for s5cmd
---
.../java/org/apache/flink/fs/s3/common/FlinkS3FileSystem.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/FlinkS3FileSystem.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/FlinkS3FileSystem.java
index af42482dcba..c8888690c5e 100644
--- a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/FlinkS3FileSystem.java
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/FlinkS3FileSystem.java
@@ -81,7 +81,7 @@ public class FlinkS3FileSystem extends HadoopFileSystem
implements EntropyInjectingFileSystem, PathsCopyingFileSystem {
private static final Logger LOG = LoggerFactory.getLogger(FlinkS3FileSystem.class);
- private static final long PROCESS_KILL_SLEEP_TIME_MS = 50L;
+ private static final long PROCESS_KILL_SLEEP_TIME_MS = 1000L;
@Nullable private final String entropyInjectionKey;
@@ -361,8 +361,9 @@ public class FlinkS3FileSystem extends HadoopFileSystem
})) {
exitCode = wizard.waitFor();
} catch (InterruptedException e) {
- Thread.currentThread().interrupt();
destroyProcess(wizard);
+ // restore interruption after destroyProcess, to let destroyProcess do some sleeps
+ Thread.currentThread().interrupt();
throw new IOException(createSpellErrorMessage(exitCode, outScrolls, artefacts), e);
} catch (IOException e) {
destroyProcess(wizard);
@@ -379,7 +380,6 @@ public class FlinkS3FileSystem extends HadoopFileSystem
}
private static void destroyProcess(Process processToDestroy) {
-
LOG.info("Destroying s5cmd copy process.");
processToDestroy.destroy();
@@ -401,7 +401,7 @@ public class FlinkS3FileSystem extends HadoopFileSystem
sleepForProcessTermination(processToDestroy);
if (processToDestroy.isAlive()) {
- LOG.warn("Could not destroy s5cmd copy process.");
+ LOG.warn("Could not destroy s5cmd copy process [pid={}].", processToDestroy.pid());
}
}