You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2020/01/10 13:48:39 UTC
[flink] 01/02: [FLINK-15319][e2e] Support configurable download
retries/timeout
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c3e0945f76ef2f18dbd5caaee062b383fe2e7d2a
Author: huzheng <op...@gmail.com>
AuthorDate: Fri Dec 20 11:06:32 2019 +0800
[FLINK-15319][e2e] Support configurable download retries/timeout
---
.../apache/flink/tests/util/AutoClosableProcess.java | 20 ++++++++++++++++++++
.../tests/util/cache/AbstractDownloadCache.java | 18 +++++++++++++++++-
2 files changed, 37 insertions(+), 1 deletion(-)
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/AutoClosableProcess.java b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/AutoClosableProcess.java
index 2d08bd8..8ed1a52 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/AutoClosableProcess.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/AutoClosableProcess.java
@@ -18,6 +18,7 @@
package org.apache.flink.tests.util;
+import org.apache.flink.api.common.time.Deadline;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -115,6 +116,25 @@ public class AutoClosableProcess implements AutoCloseable {
}
}
+ public void runBlockingWithRetry(final int maxRetries, final Duration attemptTimeout, final Duration globalTimeout) throws IOException {
+ int retries = 0;
+ final Deadline globalDeadline = Deadline.fromNow(globalTimeout);
+
+ while (true) {
+ try {
+ runBlocking(attemptTimeout);
+ break;
+ } catch (Exception e) {
+ if (++retries > maxRetries || !globalDeadline.hasTimeLeft()) {
+ String errMsg = String.format(
+ "Process (%s) exceeded timeout (%s) or number of retries (%s).",
+ Arrays.toString(commands), maxRetries, globalTimeout.toMillis());
+ throw new IOException(errMsg, e);
+ }
+ }
+ }
+ }
+
public AutoClosableProcess runNonBlocking() throws IOException {
return new AutoClosableProcess(createProcess(commands, stdoutProcessor, stderrProcessor));
}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/cache/AbstractDownloadCache.java b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/cache/AbstractDownloadCache.java
index 7d5e913..4250f1d 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/cache/AbstractDownloadCache.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/cache/AbstractDownloadCache.java
@@ -21,6 +21,8 @@ package org.apache.flink.tests.util.cache;
import org.apache.flink.tests.util.AutoClosableProcess;
import org.apache.flink.tests.util.CommandLineWrapper;
import org.apache.flink.tests.util.TestUtils;
+import org.apache.flink.tests.util.parameters.ParameterProperty;
+import org.apache.flink.util.TimeUtils;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
@@ -43,14 +45,28 @@ abstract class AbstractDownloadCache implements DownloadCache {
protected final Logger log = LoggerFactory.getLogger(getClass());
+ private static final ParameterProperty<Duration> DOWNLOAD_ATTEMPT_TIMEOUT = new ParameterProperty<>("cache-download-attempt-timeout", TimeUtils::parseDuration);
+ private static final ParameterProperty<Duration> DOWNLOAD_GLOBAL_TIMEOUT = new ParameterProperty<>("cache-download-global-timeout", TimeUtils::parseDuration);
+ private static final ParameterProperty<Integer> DOWNLOAD_MAX_RETRIES = new ParameterProperty<>("cache-download-max-retries", Integer::parseInt);
+
private final Path tmpDir;
private final Path downloadsDir;
private final Path cacheFilesDir;
+ private final Duration downloadAttemptTimeout;
+ private final Duration downloadGlobalTimeout;
+ private final int downloadMaxRetries;
+
AbstractDownloadCache(final Path tmpDir) {
this.tmpDir = tmpDir;
this.downloadsDir = tmpDir.resolve("downloads");
this.cacheFilesDir = tmpDir.resolve("cachefiles");
+
+ this.downloadAttemptTimeout = DOWNLOAD_ATTEMPT_TIMEOUT.get(Duration.ofMinutes(2));
+ this.downloadGlobalTimeout = DOWNLOAD_GLOBAL_TIMEOUT.get(Duration.ofMinutes(2));
+ this.downloadMaxRetries = DOWNLOAD_MAX_RETRIES.get(1);
+
+ log.info("Download configuration: maxAttempts={}, attemptTimeout={}, globalTimeout={}", downloadMaxRetries, downloadAttemptTimeout, downloadGlobalTimeout);
}
@Override
@@ -108,7 +124,7 @@ abstract class AbstractDownloadCache implements DownloadCache {
CommandLineWrapper.wget(url)
.targetDir(scopedDownloadDir)
.build())
- .runBlocking(Duration.ofMinutes(2));
+ .runBlockingWithRetry(downloadMaxRetries, downloadAttemptTimeout, downloadGlobalTimeout);
final Path download;
try (Stream<Path> files = Files.list(scopedDownloadDir)) {