[FLINK-12670][runtime] Implement FailureRateRestartBackoffTimeStrategy

docs/_includes/generated/restart_backoff_time_strategy_configuration.html

@@ -0,0 +1,26 @@
+<table class="table table-bordered">
+    <thead>
+        <tr>
+            <th class="text-left" style="width: 20%">Key</th>
+            <th class="text-left" style="width: 15%">Default</th>
+            <th class="text-left" style="width: 65%">Description</th>
+        </tr>
+    </thead>
+    <tbody>
+        <tr>
+            <td><h5>restart-backoff-time-strategy.failure-rate.backoff-time</h5></td>
+            <td style="word-wrap: break-word;">0</td>
+            <td>Backoff time in milliseconds between two consecutive restart attempts.</td>
+        </tr>
+        <tr>
+            <td><h5>restart-backoff-time-strategy.failure-rate.failure-rate-interval</h5></td>
+            <td style="word-wrap: break-word;">60000</td>
+            <td>Time interval in milliseconds for measuring failure rate.</td>
+        </tr>
+        <tr>
+            <td><h5>restart-backoff-time-strategy.failure-rate.max-failures-per-interval</h5></td>
+            <td style="word-wrap: break-word;">1</td>
+            <td>Maximum number of failures in given time interval before failing a job.</td>
+        </tr>
+    </tbody>
+</table>

flink-core/src/main/java/org/apache/flink/configuration/RestartBackoffTimeStrategyOptions.java

@@ -0,0 +1,56 @@
+/*
+ * 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.configuration;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * Configuration options for the RestartBackoffTimeStrategy.
+ */
+@PublicEvolving
+public class RestartBackoffTimeStrategyOptions {
+	/**
+	 * Maximum number of failures in given time interval {@link #RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_FAILURE_RATE_INTERVAL}
+	 * before failing a job in FailureRateRestartBackoffTimeStrategy.
+	 */
+	@PublicEvolving
+	public static final ConfigOption<Integer> RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_MAX_FAILURES_PER_INTERVAL = ConfigOptions
+		.key("restart-backoff-time-strategy.failure-rate.max-failures-per-interval")
+		.defaultValue(1)
+		.withDescription("Maximum number of failures in given time interval before failing a job.");
+
+	/**
+	 * Time interval in which greater amount of failures than {@link #RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_MAX_FAILURES_PER_INTERVAL}
+	 * causes job fail in FailureRateRestartBackoffTimeStrategy.
+	 */
+	@PublicEvolving
+	public static final ConfigOption<Long> RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_FAILURE_RATE_INTERVAL = ConfigOptions
+		.key("restart-backoff-time-strategy.failure-rate.failure-rate-interval")
+		.defaultValue(60_000L)
+		.withDescription("Time interval in milliseconds for measuring failure rate.");
+
+	/**
+	 * Backoff time between two consecutive restart attempts in FailureRateRestartBackoffTimeStrategy.
+	 */
+	@PublicEvolving
+	public static final ConfigOption<Long> RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_FAILURE_RATE_BACKOFF_TIME = ConfigOptions
+		.key("restart-backoff-time-strategy.failure-rate.backoff-time")
+		.defaultValue(0L)
+		.withDescription("Backoff time in milliseconds between two consecutive restart attempts.");
+}

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/FailureRateRestartBackoffTimeStrategy.java

@@ -0,0 +1,149 @@
+/*
+ * 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 org.apache.flink.runtime.util.clock.Clock;
+import org.apache.flink.runtime.util.clock.SystemClock;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+import static org.apache.flink.configuration.RestartBackoffTimeStrategyOptions.RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_FAILURE_RATE_BACKOFF_TIME;
+import static org.apache.flink.configuration.RestartBackoffTimeStrategyOptions.RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_FAILURE_RATE_INTERVAL;
+import static org.apache.flink.configuration.RestartBackoffTimeStrategyOptions.RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_MAX_FAILURES_PER_INTERVAL;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Restart strategy which can restart when failure rate is not exceeded.
+ */
+public class FailureRateRestartBackoffTimeStrategy implements RestartBackoffTimeStrategy {
+
+	private final long failuresIntervalMS;
+
+	private final long backoffTimeMS;
+
+	private final int maxFailuresPerInterval;
+
+	private final Deque<Long> failureTimestamps;
+
+	private final String strategyString;
+
+	private final Clock clock;
+
+	FailureRateRestartBackoffTimeStrategy(Clock clock, int maxFailuresPerInterval, long failuresIntervalMS, long backoffTimeMS) {
+
+		checkArgument(maxFailuresPerInterval > 0, "Maximum number of restart attempts per time unit must be greater than 0.");
+		checkArgument(failuresIntervalMS > 0, "Failures interval must be greater than 0 ms.");
+		checkArgument(backoffTimeMS >= 0, "Backoff time must be at least 0 ms.");
+
+		this.failuresIntervalMS = failuresIntervalMS;
+		this.backoffTimeMS = backoffTimeMS;
+		this.maxFailuresPerInterval = maxFailuresPerInterval;
+		this.failureTimestamps = new ArrayDeque<>(maxFailuresPerInterval);
+		this.strategyString = generateStrategyString();
+		this.clock = checkNotNull(clock);
+	}
+
+	@Override
+	public boolean canRestart() {
+		if (isFailureTimestampsQueueFull()) {
+			Long now = clock.absoluteTimeMillis();
+			Long earliestFailure = failureTimestamps.peek();
+
+			return (now - earliestFailure) > failuresIntervalMS;
+		} else {
+			return true;
+		}
+	}
+
+	@Override
+	public long getBackoffTime() {
+		return backoffTimeMS;
+	}
+
+	@Override
+	public void notifyFailure(Throwable cause) {
+		if (isFailureTimestampsQueueFull()) {
+			failureTimestamps.remove();
+		}
+		failureTimestamps.add(clock.absoluteTimeMillis());
+	}
+
+	@Override
+	public String toString() {
+		return strategyString;
+	}
+
+	private boolean isFailureTimestampsQueueFull() {
+		return failureTimestamps.size() >= maxFailuresPerInterval;
+	}
+
+	private String generateStrategyString() {
+		StringBuilder str = new StringBuilder("FailureRateRestartBackoffTimeStrategy(");
+		str.append("FailureRateRestartBackoffTimeStrategy(failuresIntervalMS=");
+		str.append(failuresIntervalMS);
+		str.append(",backoffTimeMS=");
+		str.append(backoffTimeMS);
+		str.append(",maxFailuresPerInterval=");
+		str.append(maxFailuresPerInterval);
+		str.append(")");
+
+		return str.toString();
+	}
+
+	public static FailureRateRestartBackoffTimeStrategyFactory createFactory(final Configuration configuration) {
+		return new FailureRateRestartBackoffTimeStrategyFactory(
+				configuration.getInteger(RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_MAX_FAILURES_PER_INTERVAL),
+				configuration.getLong(RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_FAILURE_RATE_INTERVAL),
+				configuration.getLong(RESTART_BACKOFF_TIME_STRATEGY_FAILURE_RATE_FAILURE_RATE_BACKOFF_TIME));
+	}
+
+	/**
+	 * The factory for creating {@link FailureRateRestartBackoffTimeStrategy}.
+	 */
+	public static class FailureRateRestartBackoffTimeStrategyFactory implements RestartBackoffTimeStrategy.Factory {
+
+		private final int maxFailuresPerInterval;
+
+		private final long failuresIntervalMS;
+
+		private final long backoffTimeMS;
+
+		public FailureRateRestartBackoffTimeStrategyFactory(
+				int maxFailuresPerInterval,
+				long failuresIntervalMS,
+				long backoffTimeMS) {
+
+			this.maxFailuresPerInterval = maxFailuresPerInterval;
+			this.failuresIntervalMS = failuresIntervalMS;
+			this.backoffTimeMS = backoffTimeMS;
+		}
+
+		@Override
+		public RestartBackoffTimeStrategy create() {
+			return new FailureRateRestartBackoffTimeStrategy(
+				SystemClock.getInstance(),
+				maxFailuresPerInterval,
+				failuresIntervalMS,
+				backoffTimeMS);
+		}
+	}
+}

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/FailureRateRestartBackoffTimeStrategyTest.java

@@ -0,0 +1,83 @@
+/*
+ * 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.runtime.util.clock.ManualClock;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Unit tests for {@link FailureRateRestartBackoffTimeStrategy}.
+ */
+public class FailureRateRestartBackoffTimeStrategyTest extends TestLogger {
+
+	private final Exception failure = new Exception();
+
+	@Test
+	public void testManyFailuresWithinRate() {
+		final int numFailures = 3;
+		final long intervalMS = 1L;
+
+		ManualClock clock = new ManualClock();
+
+		final FailureRateRestartBackoffTimeStrategy restartStrategy =
+			new FailureRateRestartBackoffTimeStrategy(clock, 1, intervalMS, 0);
+
+		for (int failuresLeft = numFailures; failuresLeft > 0; failuresLeft--) {
+			assertTrue(restartStrategy.canRestart());
+			restartStrategy.notifyFailure(failure);
+			clock.advanceTime(intervalMS + 1, TimeUnit.MILLISECONDS);
+		}
+
+		assertTrue(restartStrategy.canRestart());
+	}
+
+	@Test
+	public void testFailuresExceedingRate() {
+		final int numFailures = 3;
+		final long intervalMS = 10_000L;
+
+		final FailureRateRestartBackoffTimeStrategy restartStrategy =
+			new FailureRateRestartBackoffTimeStrategy(new ManualClock(), numFailures, intervalMS, 0);
+
+		for (int failuresLeft = numFailures; failuresLeft > 0; failuresLeft--) {
+			assertTrue(restartStrategy.canRestart());
+			restartStrategy.notifyFailure(failure);
+		}
+
+		assertFalse(restartStrategy.canRestart());
+	}
+
+	@Test
+	public void testBackoffTime() {
+		final long backoffTimeMS = 10_000L;
+
+		final FailureRateRestartBackoffTimeStrategy restartStrategy =
+			new FailureRateRestartBackoffTimeStrategy(new ManualClock(), 1, 1, backoffTimeMS);
+
+		assertEquals(backoffTimeMS, restartStrategy.getBackoffTime());
+	}
+}