[FLINK-12876][runtime] Add an adapter of region failover NG for legacy scheduler #8922

GJL · 2019-06-28T07:25:45Z

What is the purpose of the change

Based on #8851

Brief change log

See [FLINK-12876] [runtime] Add an adapter of region failover NG for legacy scheduler #8851

Verifying this change

This change is already covered by existing tests, such as RegionFailoverITCase.

This change added tests and can be verified as follows:

Added unit tests for AdaptedRestartPipelinedRegionStrategyNG

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2019-06-28T07:27:38Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

GJL · 2019-06-28T07:28:47Z

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java

+		final List<ExecutionVertex> executionVertices = StreamSupport
+			.stream(getAllExecutionVertices().spliterator(), false)
+			.collect(Collectors.toList());
+		return SchedulingUtils.scheduleLazy(executionVertices, this);


SchedulingUtils could accept Iterable

zentol · 2019-06-28T08:39:39Z

...e/src/main/java/org/apache/flink/runtime/executiongraph/failover/FailoverStrategyLoader.java

 					return new RestartAllStrategy.Factory();

 				case PIPELINED_REGION_RESTART_STRATEGY_NAME:
+				return new AdaptedRestartPipelinedRegionStrategyNG.Factory();


indentation

zentol · 2019-06-28T08:51:11Z

...ntime/executiongraph/AdaptedRestartPipelinedRegionStrategyNGAbortPendingCheckpointsTest.java

+	}
+
+	private static void assertNoPendingCheckpoints(final CheckpointCoordinator checkpointCoordinator) {
+		assertThat(checkpointCoordinator.getPendingCheckpoints().keySet(), hasSize(0));


maybe also check that CheckpointCoordinator#getSuccessfulCheckpoints() is empty

I think it's not needed since we are not acknowledging any checkpoints from the executions.

zentol · 2019-06-28T09:16:57Z

...apache/flink/runtime/executiongraph/AdaptedRestartPipelinedRegionStrategyNGFailoverTest.java

+		assertThat(failoverStrategy.getLastTasksToCancel(), containsInAnyOrder(
+			new ExecutionVertexID(ev11.getJobvertexId(), 0),
+			new ExecutionVertexID(ev21.getJobvertexId(), 0),
+			new ExecutionVertexID(ev21.getJobvertexId(), 1)));


would be easier to understand if this were ev22.getJobvertexId

true, I am waiting for the PR that enables me to write ev22.getID()

Fixed. Commit cherry picked from #8804.

zentol · 2019-06-28T09:25:25Z

...apache/flink/runtime/executiongraph/AdaptedRestartPipelinedRegionStrategyNGFailoverTest.java

+		ev12.getCurrentExecutionAttempt().markFinished();
+
+		// force FINISHED ev11 to fail to reset its partition
+		strategy.onTaskFailure(ev11.getCurrentExecutionAttempt(), new FlinkException("Fail for testing"));


I don't believe this scenario to be possible in production. FAILED to FINISHED is not a valid state transition. Could we not fail one of the consumers instead?

Let's delete the test. The adapter is not calling IntermediateResultPartition#resetForNewExecution()

zentol · 2019-06-28T09:40:06Z

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java

+		final List<ExecutionVertex> executionVertices = StreamSupport
+			.stream(getAllExecutionVertices().spliterator(), false)
+			.collect(Collectors.toList());
+		return SchedulingUtils.scheduleLazy(executionVertices, this);


zhuzhurk · 2019-06-28T09:14:53Z

...apache/flink/runtime/executiongraph/AdaptedRestartPipelinedRegionStrategyNGFailoverTest.java

+
+		private final SlotProvider delegate;
+
+


duplicated empty line

good catch, fixed

zhuzhurk · 2019-06-28T09:34:13Z

...nk/runtime/executiongraph/AdaptedRestartPipelinedRegionStrategyNGConcurrentFailoverTest.java

+
+		final FailoverStrategy failoverStrategy = executionGraph.getFailoverStrategy();
+		failoverStrategy.onTaskFailure(firstVertex.getCurrentExecutionAttempt(), new Exception("Test Exception"));
+


manualMainThreadExecutor.triggerAll() is needed here.

Wouldn't hurt but not needed since we are only interested in the state transition to ExecutionState.CANCELED.

The vertex directly transitions to CANCELED because it was no scheduled yet and the graph gets cancelled. It does not prove whether the local failover is skipped. (Even though a local failover is not skipped, the recovery action will be queued in main thread executor and not able to change the vertex state out from CANCELED).

If we want to verify "skip local failover if the job status is not Running", how about trigger task failure directly without graph.cancel()? In this case, the local failover will be skipped since the graph is in CREATED state, and the vertex will remain in CREATED state.

I'll add triggerAll() since it is a more realistic scenario, and it is less invasive (createExecutionGraph() already schedules the ExecutionGraph).

zhuzhurk · 2019-06-28T09:48:05Z

...apache/flink/runtime/executiongraph/AdaptedRestartPipelinedRegionStrategyNGFailoverTest.java


 		// trigger downstream regions to schedule
-		testingMainThreadExecutor.execute(() -> {
+		componentMainThreadExecutor.execute(() -> {


can invoke markFinished directly.

Thanks, fixed it.

GJL · 2019-06-29T15:23:48Z

rebased to master

zhuzhurk · 2019-07-01T02:28:21Z

...e/src/main/java/org/apache/flink/runtime/executiongraph/failover/FailoverStrategyLoader.java

+	/** Config name for the {@link AdaptedRestartPipelinedRegionStrategyNG} */
 	public static final String PIPELINED_REGION_RESTART_STRATEGY_NAME = "region";

+	/** Config name for the {@link RestartPipelinedRegionStrategy} */


A "." is needed at the end of the statement.
Also for the other 3 config name comments.

tillrohrmann

Thanks for creating this PR @GJL and @zhuzhurk. I've looked at the functional changes and they look good to me. I had some minor comments which we could address.

I will be looking now at the tests in detail.

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/SchedulingUtils.java

tillrohrmann · 2019-07-02T14:18:52Z

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/SchedulingUtils.java

+	}
+
+	/**
+	 * Schedule vertices lazy. That means all vertices will be scheduled at once.


flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/SchedulingUtils.java

...rg/apache/flink/runtime/executiongraph/failover/AdaptedRestartPipelinedRegionStrategyNG.java

tillrohrmann

I finished my pass over the tests. The changes look good to me. Thanks a lot for the very good work @GJL and @zhuzhurk as well as for the review @zentol. I really like how you tested the code. One can really see that you put some good thoughts into it.

+1 for merging.

tillrohrmann · 2019-07-02T19:54:35Z

...ache/flink/runtime/executiongraph/ScheduledExecutorToComponentMainThreadExecutorAdapter.java

+/**
+ * Adapts {@link ScheduledExecutor} to {@link ComponentMainThreadExecutor}.
+ */
+public class ScheduledExecutorToComponentMainThreadExecutorAdapter implements ComponentMainThreadExecutor {


I think this class duplicates ComponentMainThreadExecutorServiceAdapter. A related class is TestingComponentMainThreadExecutorServiceAdapter.

I don't think so.

ComponentMainThreadExecutorServiceAdapter adapts ScheduledExecutorService. ScheduledExecutor is not a ScheduledExecutorService. There is no adapter from ScheduledExecutor to ScheduledExecutorService.

But there should be ScheduledExecutorServiceAdapter for ScheduledExecutorService -> ScheduledExecutor, right?

I made ComponentMainThreadExecutorServiceAdapter accept ScheduledExecutor and deleted my adapter. However, there are more pending clean ups that require a separate PR – for example, it does not make sense that TestingComponentMainThreadExecutorServiceAdapter extends ComponentMainThreadExecutorServiceAdapter.

GJL · 2019-07-03T09:15:39Z

rebased to master

… from ManuallyTriggeredScheduledExecutor

GJL · 2019-07-03T17:54:56Z

Rebased to master and squashed commits. Merging as soon as build is green.

…pt ScheduledExecutor

…gacy scheduling Implement adapter (AdaptedRestartPipelinedRegionStrategyNG) that adapts org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionStrategy to the legacy failover strategy interface (org.apache.flink.runtime.executiongraph.failover.FailoverStrategy). The new AdaptedRestartPipelinedRegionStrategyNG is chosen if config option jobmanager.execution.failover-strategy is set to "region". The legacy behavior can be enabled by setting the config option to "region-legacy".

rmetzger added the review=description? label Jun 28, 2019

GJL commented Jun 28, 2019

View reviewed changes

rmetzger added the component=Runtime/Coordination label Jun 28, 2019

zentol reviewed Jun 28, 2019

View reviewed changes

zhuzhurk reviewed Jun 28, 2019

View reviewed changes

GJL force-pushed the FLINK-12876 branch from d4e3eff to 13f43ec Compare June 29, 2019 15:23

zhuzhurk reviewed Jul 1, 2019

View reviewed changes

GJL mentioned this pull request Jul 1, 2019

[FLINK-12876] [runtime] Add an adapter of region failover NG for legacy scheduler #8851

Closed

zentol self-assigned this Jul 2, 2019

tillrohrmann reviewed Jul 2, 2019

View reviewed changes

tillrohrmann approved these changes Jul 2, 2019

View reviewed changes

GJL force-pushed the FLINK-12876 branch from 725428b to f29ff79 Compare July 3, 2019 08:57

azagrebin mentioned this pull request Jul 3, 2019

[FLINK-13019][coordination][tests] IT test for fine-grained recovery #8971

Merged

zentol mentioned this pull request Jul 3, 2019

[FLINK-13060][coordination] Respect restart constraints in new RegionFailover strategy #8972

Merged

GJL force-pushed the FLINK-12876 branch from b9366bf to fc5d4cc Compare July 3, 2019 17:53

GJL added 2 commits July 3, 2019 19:54

[hotfix][runtime] Fix checkstyle violations in FailoverStrategyLoader

0074bef

[hotfix][runtime, tests] Remove ComponentMainThreadExecutor interface…

5aba0b6

… from ManuallyTriggeredScheduledExecutor

GJL force-pushed the FLINK-12876 branch from fc5d4cc to d351ced Compare July 3, 2019 17:54

GJL added 2 commits July 3, 2019 20:19

[hotfix][runtime] Make ComponentMainThreadExecutorServiceAdapter acce…

498f498

…pt ScheduledExecutor

GJL force-pushed the FLINK-12876 branch from d351ced to 557195c Compare July 3, 2019 18:22

asfgit closed this in 6a682ff Jul 3, 2019

GJL deleted the FLINK-12876 branch September 3, 2019 09:55


		final FailoverStrategy failoverStrategy = executionGraph.getFailoverStrategy();
		failoverStrategy.onTaskFailure(firstVertex.getCurrentExecutionAttempt(), new Exception("Test Exception"));

[FLINK-12876][runtime] Add an adapter of region failover NG for legacy scheduler #8922

[FLINK-12876][runtime] Add an adapter of region failover NG for legacy scheduler #8922

Uh oh!

Conversation

GJL commented Jun 28, 2019

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Jun 28, 2019

Review Progress

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL Jun 28, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhuzhurk Jun 28, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL commented Jun 29, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tillrohrmann left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

GJL Jun 28, 2019 •

edited

Loading

zhuzhurk Jun 28, 2019 •

edited

Loading

tillrohrmann left a comment •

edited

Loading