[FLINK-19693][runtime] Downstream Failover for Approximate Local Recovery

flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java

@@ -82,6 +82,9 @@ public class JobGraph implements Serializable {
 	/** The mode in which the job is scheduled. */
 	private ScheduleMode scheduleMode = ScheduleMode.LAZY_FROM_SOURCES;
 
+	/** Whether approximate local recovery is enabled. This flag will be removed together with legacy scheduling strategies. */
+	private boolean approximateLocalRecovery = false;
+
 	// --- checkpointing ---
 
 	/** Job specific execution config. */
@@ -231,6 +234,14 @@ public ScheduleMode getScheduleMode() {
 		return scheduleMode;
 	}
 
+	public void enableApproximateLocalRecovery(boolean enabled) {
+		this.approximateLocalRecovery = enabled;
+	}
+
+	public boolean isApproximateLocalRecoveryEnabled() {
+		return approximateLocalRecovery;
+	}
+
 	/**
 	 * Sets the savepoint restore settings.
 	 * @param settings The savepoint restore settings.

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultSchedulerComponents.java

@@ -45,6 +45,8 @@
 
 import java.util.function.Consumer;
 
+import static org.apache.flink.util.Preconditions.checkArgument;
+
 /**
  * Components to create a {@link DefaultScheduler} which depends on the
  * configured {@link JobManagerOptions#SCHEDULING_STRATEGY}.
@@ -82,19 +84,26 @@ ExecutionSlotAllocatorFactory getAllocatorFactory() {
 
 	static DefaultSchedulerComponents createSchedulerComponents(
 			final ScheduleMode scheduleMode,
+			final boolean isApproximateLocalRecoveryEnabled,
 			final Configuration jobMasterConfiguration,
 			final SlotPool slotPool,
 			final Time slotRequestTimeout) {
 
 		final String schedulingStrategy = jobMasterConfiguration.getString(JobManagerOptions.SCHEDULING_STRATEGY);
 		switch (schedulingStrategy) {
 			case PIPELINED_REGION_SCHEDULING:
+				checkArgument(
+					!isApproximateLocalRecoveryEnabled,
+					"Approximate local recovery can not be used together with PipelinedRegionScheduler for now! " +
+						"Please set JobManagerOptions.SCHEDULING_STRATEGY to legacy.");
 				return createPipelinedRegionSchedulerComponents(
 					scheduleMode,
 					jobMasterConfiguration,
 					slotPool,
 					slotRequestTimeout);
 			case LEGACY_SCHEDULING:
+				checkArgument(!isApproximateLocalRecoveryEnabled || !scheduleMode.allowLazyDeployment(),
+					"Approximate local recovery can only be used together with EAGER schedule mode!");
 				return createLegacySchedulerComponents(
 					scheduleMode,
 					jobMasterConfiguration,

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultSchedulerFactory.java

@@ -69,6 +69,7 @@ public SchedulerNG createInstance(
 
 		final DefaultSchedulerComponents schedulerComponents = createSchedulerComponents(
 			jobGraph.getScheduleMode(),
+			jobGraph.isApproximateLocalRecoveryEnabled(),
 			jobMasterConfiguration,
 			slotPool,
 			slotRequestTimeout);

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerComponentsFactoryTest.java

@@ -26,14 +26,19 @@
 import org.apache.flink.runtime.jobgraph.ScheduleMode;
 import org.apache.flink.runtime.jobmaster.slotpool.SlotPool;
 import org.apache.flink.runtime.jobmaster.slotpool.TestingSlotPoolImpl;
+import org.apache.flink.runtime.scheduler.strategy.EagerSchedulingStrategy;
 import org.apache.flink.runtime.scheduler.strategy.LazyFromSourcesSchedulingStrategy;
 import org.apache.flink.runtime.scheduler.strategy.PipelinedRegionSchedulingStrategy;
 import org.apache.flink.util.TestLogger;
 
 import org.junit.Test;
 
+import static org.apache.flink.runtime.jobgraph.ScheduleMode.EAGER;
+import static org.apache.flink.runtime.jobgraph.ScheduleMode.LAZY_FROM_SOURCES_WITH_BATCH_SLOT_REQUEST;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 /**
  * Tests for the factory method {@link DefaultSchedulerComponents#createSchedulerComponents(
@@ -67,9 +72,56 @@ public void testCreatingPipelinedRegionSchedulingStrategyFactoryByDefault() {
 		assertThat(components.getSchedulingStrategyFactory(), instanceOf(PipelinedRegionSchedulingStrategy.Factory.class));
 	}
 
+	@Test
+	public void testCreatingPipelinedRegionSchedulingStrategyFactoryWithApproximateLocalRecovery() {
+		final Configuration configuration = new Configuration();
+		configuration.setString(JobManagerOptions.SCHEDULING_STRATEGY, "region");
+
+		try {
+			createSchedulerComponents(configuration, true, EAGER);
+			fail("expected failure");
+		} catch (IllegalArgumentException e) {
+			assertTrue(
+				e.getMessage()
+					.contains("Approximate local recovery can not be used together with PipelinedRegionScheduler for now"));
+		}
+	}
+
+	@Test
+	public void testCreatingLegacySchedulingStrategyFactoryWithApproximateLocalRecoveryInLazyMode() {
+		final Configuration configuration = new Configuration();
+		configuration.setString(JobManagerOptions.SCHEDULING_STRATEGY, "legacy");
+
+		try {
+			createSchedulerComponents(configuration, true, LAZY_FROM_SOURCES_WITH_BATCH_SLOT_REQUEST);
+			fail("expected failure");
+		} catch (IllegalArgumentException e) {
+			assertTrue(
+				e.getMessage()
+					.contains("Approximate local recovery can only be used together with EAGER schedule mode"));
+		}
+	}
+
+	@Test
+	public void testCreatingLegacySchedulingStrategyFactoryWithApproximateLocalRecoveryInEagerMode() {
+		final Configuration configuration = new Configuration();
+		configuration.setString(JobManagerOptions.SCHEDULING_STRATEGY, "legacy");
+
+		final DefaultSchedulerComponents components = createSchedulerComponents(configuration, true, EAGER);
+		assertThat(components.getSchedulingStrategyFactory(), instanceOf(EagerSchedulingStrategy.Factory.class));
+	}
+
 	private static DefaultSchedulerComponents createSchedulerComponents(final Configuration configuration) {
+		return createSchedulerComponents(configuration, false, LAZY_FROM_SOURCES_WITH_BATCH_SLOT_REQUEST);
+	}
+
+	private static DefaultSchedulerComponents createSchedulerComponents(
+			final Configuration configuration,
+			boolean iApproximateLocalRecoveryEnabled,
+			ScheduleMode scheduleMode) {
 		return DefaultSchedulerComponents.createSchedulerComponents(
-			ScheduleMode.LAZY_FROM_SOURCES_WITH_BATCH_SLOT_REQUEST,
+			scheduleMode,
+			iApproximateLocalRecoveryEnabled,
 			configuration,
 			new TestingSlotPoolImpl(new JobID()),
 			Time.milliseconds(10L));

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.streaming.api.environment;
 
+import org.apache.flink.annotation.Experimental;
 import org.apache.flink.annotation.Public;
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.JobStatus;
@@ -84,6 +85,9 @@ public class CheckpointConfig implements java.io.Serializable {
 
 	private long alignmentTimeout = ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT.defaultValue().toMillis();
 
+	/** Flag to enable approximate local recovery. */
+	private boolean approximateLocalRecovery;
+
 	/** Cleanup behaviour for persistent checkpoints. */
 	private ExternalizedCheckpointCleanup externalizedCheckpointCleanup;
 
@@ -122,6 +126,7 @@ public CheckpointConfig(final CheckpointConfig checkpointConfig) {
 		this.tolerableCheckpointFailureNumber = checkpointConfig.tolerableCheckpointFailureNumber;
 		this.unalignedCheckpointsEnabled = checkpointConfig.isUnalignedCheckpointsEnabled();
 		this.alignmentTimeout = checkpointConfig.alignmentTimeout;
+		this.approximateLocalRecovery = checkpointConfig.isApproximateLocalRecoveryEnabled();
 		this.externalizedCheckpointCleanup = checkpointConfig.externalizedCheckpointCleanup;
 		this.forceCheckpointing = checkpointConfig.forceCheckpointing;
 		this.forceUnalignedCheckpoints = checkpointConfig.forceUnalignedCheckpoints;
@@ -504,6 +509,35 @@ public long getAlignmentTimeout() {
 		return alignmentTimeout;
 	}
 
+	/**
+	 * Returns whether approximate local recovery is enabled.
+	 *
+	 * @return <code>true</code> if approximate local recovery is enabled.
+	 */
+	@Experimental
+	public boolean isApproximateLocalRecoveryEnabled() {
+		return approximateLocalRecovery;
+	}
+
+	/**
+	 * Enables the approximate local recovery mode.
+	 *
+	 * <p>In this recovery mode, when a task fails, the entire downstream of the tasks (including the failed task) restart.
+	 *
+	 * <p>Notice that
+	 * 1. Approximate recovery may lead to data loss. The amount of data which leads the failed task
+	 * from the state of the last completed checkpoint to the state when the task fails is lost.
+	 * 2. In the next version, we will support restarting the set of failed set of tasks only.
+	 * In this version, we only support downstream restarts when a task fails.
+	 * 3. It is only an internal feature for now.
+	 *
+	 * @param enabled Flag to indicate whether approximate local recovery is enabled .
+	 */
+	@Experimental
+	public void enableApproximateLocalRecovery(boolean enabled) {
+		approximateLocalRecovery = enabled;
+	}
+
 	/**
 	 * Returns the cleanup behaviour for externalized checkpoints.
 	 *

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/GlobalDataExchangeMode.java

@@ -44,5 +44,8 @@ public enum GlobalDataExchangeMode {
 	POINTWISE_EDGES_PIPELINED,
 
 	/** Set all job edges {@link ResultPartitionType#PIPELINED_BOUNDED}. */
-	ALL_EDGES_PIPELINED
+	ALL_EDGES_PIPELINED,
+
+	/** Set all job edges {@link ResultPartitionType#PIPELINED_APPROXIMATE}. */
+	ALL_EDGES_PIPELINED_APPROXIMATE
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java

@@ -287,11 +287,23 @@ private void configureStreamGraph(final StreamGraph graph) {
 			setBatchStateBackendAndTimerService(graph);
 		} else {
 			graph.setStateBackend(stateBackend);
-			graph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_PIPELINED);
 			graph.setScheduleMode(ScheduleMode.EAGER);
+
+			if (checkpointConfig.isApproximateLocalRecoveryEnabled()) {
+				checkApproximateLocalRecoveryCompatibility();
+				graph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_PIPELINED_APPROXIMATE);
+			} else {
+				graph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_PIPELINED);
+			}
 		}
 	}
 
+	private void checkApproximateLocalRecoveryCompatibility() {
+		checkState(
+			!checkpointConfig.isUnalignedCheckpointsEnabled(),
+			"Approximate Local Recovery and Unaligned Checkpoint can not be used together yet");
+	}
+
 	private void setBatchStateBackendAndTimerService(StreamGraph graph) {
 		boolean useStateBackend = configuration.get(ExecutionOptions.USE_BATCH_STATE_BACKEND);
 		boolean sortInputs = configuration.get(ExecutionOptions.SORT_INPUTS);

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java

@@ -162,6 +162,7 @@ private JobGraph createJobGraph() {
 
 		// make sure that all vertices start immediately
 		jobGraph.setScheduleMode(streamGraph.getScheduleMode());
+		jobGraph.enableApproximateLocalRecovery(streamGraph.getCheckpointConfig().isApproximateLocalRecoveryEnabled());
 
 		// Generate deterministic hashes for the nodes in order to identify them across
 		// submission iff they didn't change.
@@ -743,6 +744,8 @@ private ResultPartitionType determineResultPartitionType(StreamPartitioner<?> pa
 				}
 			case ALL_EDGES_PIPELINED:
 				return ResultPartitionType.PIPELINED_BOUNDED;
+			case ALL_EDGES_PIPELINED_APPROXIMATE:
+				return ResultPartitionType.PIPELINED_APPROXIMATE;
 			default:
 				throw new RuntimeException("Unrecognized global data exchange mode " + streamGraph.getGlobalDataExchangeMode());
 		}