Repository: flink Updated Branches: refs/heads/master d57ebd063 -> e5ab4053b
[FLINK-8099] Reduce default restart delay to 1 second Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e5ab4053 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e5ab4053 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e5ab4053 Branch: refs/heads/master Commit: e5ab4053bb20342699eaa70603b4246845608a89 Parents: d57ebd0 Author: Aljoscha Krettek <[email protected]> Authored: Fri Nov 17 17:19:51 2017 +0100 Committer: Aljoscha Krettek <[email protected]> Committed: Mon Nov 20 14:11:01 2017 +0100 ---------------------------------------------------------------------- docs/ops/config.md | 4 +--- .../flink/configuration/ConfigConstants.java | 3 ++- .../restart/FixedDelayRestartStrategy.java | 21 ++++---------------- 3 files changed, 7 insertions(+), 21 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/e5ab4053/docs/ops/config.md ---------------------------------------------------------------------- diff --git a/docs/ops/config.md b/docs/ops/config.md index 55dec4c..c85ce84 100644 --- a/docs/ops/config.md +++ b/docs/ops/config.md @@ -231,9 +231,7 @@ The options are: - `restart-strategy.fixed-delay.attempts`: Number of restart attempts, used if the default restart strategy is set to "fixed-delay". Default value is 1, unless "fixed-delay" was activated by enabling checkpoints, in which case the default is `Integer.MAX_VALUE`. -- `restart-strategy.fixed-delay.delay`: Delay between restart attempts, used if the default restart strategy is set to "fixed-delay". -Default value is the `akka.ask.timeout`, unless "fixed-delay" was activated by enabling checkpoints, in which case -the default is 10s. +- `restart-strategy.fixed-delay.delay`: Delay between restart attempts, used if the default restart strategy is set to "fixed-delay". (default: `1 s`) - `restart-strategy.failure-rate.max-failures-per-interval`: Maximum number of restarts in given time interval before failing a job in "failure-rate" strategy. Default value is 1. http://git-wip-us.apache.org/repos/asf/flink/blob/e5ab4053/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java index f9e4735..fcf73b8 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java @@ -67,7 +67,8 @@ public final class ConfigConstants { * FiniteDuration notation: "1 min", "20 s" */ @PublicEvolving - public static final String RESTART_STRATEGY_FIXED_DELAY_DELAY = "restart-strategy.fixed-delay.delay"; + public static final ConfigOption<String> RESTART_STRATEGY_FIXED_DELAY_DELAY = + key("restart-strategy.fixed-delay.delay").defaultValue("1 s"); /** * Maximum number of restarts in given time interval {@link #RESTART_STRATEGY_FAILURE_RATE_FAILURE_RATE_INTERVAL} before failing a job http://git-wip-us.apache.org/repos/asf/flink/blob/e5ab4053/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategy.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategy.java index ca9626a..1916bea 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategy.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategy.java @@ -18,7 +18,6 @@ package org.apache.flink.runtime.executiongraph.restart; -import org.apache.flink.configuration.AkkaOptions; import org.apache.flink.configuration.ConfigConstants; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.concurrent.ScheduledExecutor; @@ -81,28 +80,16 @@ public class FixedDelayRestartStrategy implements RestartStrategy { public static FixedDelayRestartStrategyFactory createFactory(Configuration configuration) throws Exception { int maxAttempts = configuration.getInteger(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 1); - String timeoutString = configuration.getString( - AkkaOptions.WATCH_HEARTBEAT_INTERVAL); - - String delayString = configuration.getString( - ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, - timeoutString - ); + String delayString = configuration.getString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY); long delay; try { delay = Duration.apply(delayString).toMillis(); } catch (NumberFormatException nfe) { - if (delayString.equals(timeoutString)) { - throw new Exception("Invalid config value for " + - AkkaOptions.WATCH_HEARTBEAT_INTERVAL.key() + ": " + timeoutString + - ". Value must be a valid duration (such as '10 s' or '1 min')"); - } else { - throw new Exception("Invalid config value for " + - ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY + ": " + delayString + - ". Value must be a valid duration (such as '100 milli' or '10 s')"); - } + throw new Exception("Invalid config value for " + + ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY + ": " + delayString + + ". Value must be a valid duration (such as '100 milli' or '10 s')"); } return new FixedDelayRestartStrategyFactory(maxAttempts, delay);
