You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2019/06/14 07:20:51 UTC
[flink] 02/02: [FLINK-12669][runtime] Implement
FixedDelayRestartBackoffTimeStrategy
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 66b38acd3b82967c44450d9eec58266e1d2fb124
Author: Bo WANG <wb...@gmail.com>
AuthorDate: Wed Jun 12 11:26:09 2019 +0800
[FLINK-12669][runtime] Implement FixedDelayRestartBackoffTimeStrategy
This closes #8698.
---
...estart_backoff_time_strategy_configuration.html | 10 ++
.../RestartBackoffTimeStrategyOptions.java | 18 ++++
.../FixedDelayRestartBackoffTimeStrategy.java | 108 +++++++++++++++++++++
.../FixedDelayRestartBackoffTimeStrategyTest.java | 63 ++++++++++++
4 files changed, 199 insertions(+)
diff --git a/docs/_includes/generated/restart_backoff_time_strategy_configuration.html b/docs/_includes/generated/restart_backoff_time_strategy_configuration.html
index 0c5c2bc..33785e0 100644
--- a/docs/_includes/generated/restart_backoff_time_strategy_configuration.html
+++ b/docs/_includes/generated/restart_backoff_time_strategy_configuration.html
@@ -22,5 +22,15 @@
<td style="word-wrap: break-word;">1</td>
<td>Maximum number of failures in given time interval before failing a job.</td>
</tr>
+ <tr>
+ <td><h5>restart-backoff-time-strategy.fixed-delay.attempts</h5></td>
+ <td style="word-wrap: break-word;">2147483647</td>
+ <td>Maximum number of attempts the fixed delay restart strategy will try before failing a job.</td>
+ </tr>
+ <tr>
+ <td><h5>restart-backoff-time-strategy.fixed-delay.backoff-time</h5></td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Backoff time in milliseconds between two consecutive restart attempts.</td>
+ </tr>
</tbody>
</table>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/RestartBackoffTimeStrategyOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/RestartBackoffTimeStrategyOptions.java
index 18ebec6..7b56a33 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/RestartBackoffTimeStrategyOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/RestartBackoffTimeStrategyOptions.java
@@ -53,4 +53,22 @@ public class RestartBackoffTimeStrategyOptions {
.key("restart-backoff-time-strategy.failure-rate.backoff-time")
.defaultValue(0L)
.withDescription("Backoff time in milliseconds between two consecutive restart attempts.");
+
+ /**
+ * Maximum number of attempts the fixed delay restart strategy will try before failing a job.
+ */
+ @PublicEvolving
+ public static final ConfigOption<Integer> RESTART_BACKOFF_TIME_STRATEGY_FIXED_DELAY_ATTEMPTS = ConfigOptions
+ .key("restart-backoff-time-strategy.fixed-delay.attempts")
+ .defaultValue(Integer.MAX_VALUE)
+ .withDescription("Maximum number of attempts the fixed delay restart strategy will try before failing a job.");
+
+ /**
+ * Backoff time between two consecutive restart attempts in FixedDelayRestartBackoffTimeStrategy.
+ */
+ @PublicEvolving
+ public static final ConfigOption<Long> RESTART_BACKOFF_TIME_STRATEGY_FIXED_DELAY_BACKOFF_TIME = ConfigOptions
+ .key("restart-backoff-time-strategy.fixed-delay.backoff-time")
+ .defaultValue(0L)
+ .withDescription("Backoff time in milliseconds between two consecutive restart attempts.");
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/FixedDelayRestartBackoffTimeStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/FixedDelayRestartBackoffTimeStrategy.java
new file mode 100644
index 0000000..64dc1bf
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/FixedDelayRestartBackoffTimeStrategy.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1;
+
+import org.apache.flink.configuration.Configuration;
+
+import static org.apache.flink.configuration.RestartBackoffTimeStrategyOptions.RESTART_BACKOFF_TIME_STRATEGY_FIXED_DELAY_ATTEMPTS;
+import static org.apache.flink.configuration.RestartBackoffTimeStrategyOptions.RESTART_BACKOFF_TIME_STRATEGY_FIXED_DELAY_BACKOFF_TIME;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Restart strategy which tries to restart a fixed number of times with a fixed backoff time in between.
+ */
+public class FixedDelayRestartBackoffTimeStrategy implements RestartBackoffTimeStrategy {
+
+ private final int maxNumberRestartAttempts;
+
+ private final long backoffTimeMS;
+
+ private final String strategyString;
+
+ private int currentRestartAttempt;
+
+ FixedDelayRestartBackoffTimeStrategy(int maxNumberRestartAttempts, long backoffTimeMS) {
+ checkArgument(maxNumberRestartAttempts >= 0, "Maximum number of restart attempts must be at least 0.");
+ checkArgument(backoffTimeMS >= 0, "Backoff time between restart attempts must be at least 0 ms.");
+
+ this.maxNumberRestartAttempts = maxNumberRestartAttempts;
+ this.backoffTimeMS = backoffTimeMS;
+ this.currentRestartAttempt = 0;
+ this.strategyString = generateStrategyString();
+ }
+
+ @Override
+ public boolean canRestart() {
+ return currentRestartAttempt <= maxNumberRestartAttempts;
+ }
+
+ @Override
+ public long getBackoffTime() {
+ return backoffTimeMS;
+ }
+
+ @Override
+ public void notifyFailure(Throwable cause) {
+ currentRestartAttempt++;
+ }
+
+ @Override
+ public String toString() {
+ return strategyString;
+ }
+
+ private String generateStrategyString() {
+ StringBuilder str = new StringBuilder("FixedDelayRestartBackoffTimeStrategy(");
+ str.append("maxNumberRestartAttempts=");
+ str.append(maxNumberRestartAttempts);
+ str.append(", backoffTimeMS=");
+ str.append(backoffTimeMS);
+ str.append(")");
+
+ return str.toString();
+ }
+
+ public static FixedDelayRestartBackoffTimeStrategyFactory createFactory(final Configuration configuration) {
+ return new FixedDelayRestartBackoffTimeStrategyFactory(
+ configuration.getInteger(RESTART_BACKOFF_TIME_STRATEGY_FIXED_DELAY_ATTEMPTS),
+ configuration.getLong(RESTART_BACKOFF_TIME_STRATEGY_FIXED_DELAY_BACKOFF_TIME));
+ }
+
+ /**
+ * The factory for creating {@link FixedDelayRestartBackoffTimeStrategy}.
+ */
+ public static class FixedDelayRestartBackoffTimeStrategyFactory implements RestartBackoffTimeStrategy.Factory {
+
+ private final int maxNumberRestartAttempts;
+
+ private final long backoffTimeMS;
+
+ public FixedDelayRestartBackoffTimeStrategyFactory(int maxNumberRestartAttempts, long backoffTimeMS) {
+ this.maxNumberRestartAttempts = maxNumberRestartAttempts;
+ this.backoffTimeMS = backoffTimeMS;
+ }
+
+ @Override
+ public RestartBackoffTimeStrategy create() {
+ return new FixedDelayRestartBackoffTimeStrategy(
+ maxNumberRestartAttempts,
+ backoffTimeMS);
+ }
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/FixedDelayRestartBackoffTimeStrategyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/FixedDelayRestartBackoffTimeStrategyTest.java
new file mode 100644
index 0000000..5c47219
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/FixedDelayRestartBackoffTimeStrategyTest.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1;
+
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import static junit.framework.TestCase.assertTrue;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+/**
+ * Unit tests for {@link FixedDelayRestartBackoffTimeStrategy}.
+ */
+public class FixedDelayRestartBackoffTimeStrategyTest extends TestLogger {
+
+ private final Exception failure = new Exception();
+
+ @Test
+ public void testNumberOfRestarts() {
+ final int numberOfRestarts = 3;
+
+ final FixedDelayRestartBackoffTimeStrategy strategy =
+ new FixedDelayRestartBackoffTimeStrategy(numberOfRestarts, 0L);
+
+ for (int restartsLeft = numberOfRestarts; restartsLeft > 0; --restartsLeft) {
+ strategy.notifyFailure(failure);
+ // two calls to 'canRestart()' to make sure this is not used to maintain the counter
+ assertTrue(strategy.canRestart());
+ assertTrue(strategy.canRestart());
+ }
+
+ strategy.notifyFailure(failure);
+ assertFalse(strategy.canRestart());
+ }
+
+ @Test
+ public void testBackoffTime() {
+ final long backoffTimeMS = 10_000L;
+
+ final FixedDelayRestartBackoffTimeStrategy restartStrategy =
+ new FixedDelayRestartBackoffTimeStrategy(1, backoffTimeMS);
+
+ assertEquals(backoffTimeMS, restartStrategy.getBackoffTime());
+ }
+}