[FLINK-8176][flip6] Start SubmittedJobGraphStore in Dispatcher #5107

GJL · 2017-11-30T16:23:24Z

What is the purpose of the change

The FLIP-6 dispatcher never calls start() on its SubmittedJobGraphStore instance. Hence, when a Job is submitted (YARN session mode with HA enabled), an IllegalStateException is thrown. This pull request adds the necessary changes so that jobs can be submitted.

Brief change log

Implement SubmittedJobGraphListener interface in Dispatcher

Verifying this change

Added unit tests for new methods in Dispatcher class
Verified that jobs can be submitted in FLIP-6 YARN session mode with HA. Did not verify anything else.

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)

CC: @tillrohrmann

GJL · 2017-11-30T16:28:22Z

flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java

@@ -246,5 +258,11 @@ protected JobManagerRunner createJobManagerRunner(

 			return jobManagerRunner;
 		}
+
+		@Override
+		protected void runAsync(final Runnable runnable) {


I think if I can set Akka's CallingThreadDispatcher, this is not needed.

GJL · 2017-12-01T11:07:24Z

flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java

+		dispatcher.submitJob(jobGraph, timeout);
+
+		// pretend that other Dispatcher has removed job from submittedJobGraphStore
+		dispatcher.onRemovedJobGraph(TEST_JOB_ID);


Test is not thread-safe.

Interacting with the dispatcher via the self gateway should help.

tillrohrmann

LGTM. After changing the tests to use the DispatcherGateway we can merge it.

tillrohrmann · 2017-12-01T14:27:29Z

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java

+			try {
+				submittedJobGraph = submittedJobGraphStore.recoverJobGraph(jobId);
+			} catch (final Exception e) {
+				log.error("Could not submit job {}.", jobId, e);


Maybe "could not recover job".

Changed it.

tillrohrmann · 2017-12-01T14:36:15Z

flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java

+		dispatcher.submitJob(jobGraph, timeout);
+
+		// pretend that other Dispatcher has removed job from submittedJobGraphStore
+		dispatcher.onRemovedJobGraph(TEST_JOB_ID);


Interacting with the dispatcher via the self gateway should help.

GJL · 2017-12-04T19:20:21Z

flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java

+
+		@Override
+		void recoverJobs() {
+			if (recoverJobsEnabled.get()) {


Without this I do not see how I can verify whether a job was submitted regularly or via recoverJobs.

…bManagerHARecoveryTest

…Dispatcher Call start() on SubmittedJobGraphStore with Dispatcher as listener. To enable this, the dispatcher must implement the SubmittedJobGraphListener interface. Add simple unit tests for the new methods. Refactor DispatcherTest to remove redundancy.

…oc end with period

…rvice

…tingHighAvailabilityServices

Check if jobManagerRunner exists before submitting job. Replace JobManagerRunner mock used in tests with real instance. Do not run job graph recovery in actor main thread when job graph is recovered from SubmittedJobGraphListener#onAddedJobGraph(JobID).

tillrohrmann

Changes look good to me. Thanks for your contribution @GJL. I only had minor comments. After addressing them, this PR is good to be merged.

tillrohrmann · 2017-12-06T12:43:22Z

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java

+	@Override
+	public void onRemovedJobGraph(final JobID jobId) {
+		runAsync(() -> {
+			if (jobManagerRunners.containsKey(jobId)) {


This check should be superfluous because it will be check in removeJob as well.

tillrohrmann · 2017-12-06T12:43:46Z

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java

+				return;
+			}
+			runAsync(() -> {
+				if (!jobManagerRunners.containsKey(jobId)) {


I think this check is not strictly needed.

tillrohrmann · 2017-12-06T12:45:48Z

flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java

-		TestingFatalErrorHandler fatalErrorHandler = new TestingFatalErrorHandler();
+	@Before
+	public void setUp() throws Exception {
+		MockitoAnnotations.initMocks(this);


No longer needed.

Nice catch.

tillrohrmann · 2017-12-06T13:00:26Z

...runtime/src/test/java/org/apache/flink/runtime/testutils/InMemorySubmittedJobGraphStore.java

+	private boolean started;
+
+	@Override
+	public synchronized void start(@Nullable SubmittedJobGraphListener jobGraphListener) throws Exception {


I think we should also trigger SubmittedJobGraphListener notifications when a job is added or removed from this store to make it behave similarly to the ZooKeeperSubmittedJobGraphStore.

Good point.

Not sure if it is actually applicable. onRemovedJobGraph explicitly demands that the graph is removed by a different SubmittedJobGraphStore instance:

/** * Callback for {@link SubmittedJobGraph} instances removed by a different {@link * SubmittedJobGraphStore} instance. * * @param jobId The {@link JobID} of the removed job graph */ void onRemovedJobGraph(JobID jobId);

tillrohrmann · 2017-12-06T13:38:52Z

flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java

+	 */
+	@Test
+	public void testSubmittedJobGraphListener() throws Exception {
+		dispatcher.recoverJobsEnabled.set(false);


Why do we need this to be set to false? Recover all jobs should only be triggered in case of a leadership grant to the dispatcher which only happens at the very beginning of this test.

It is not so predictable when the main logic of recoverJobs() is scheduled because it does not run in the main thread:

/** * Recovers all jobs persisted via the submitted job graph store. */ @VisibleForTesting void recoverJobs() { log.info("Recovering all persisted jobs."); getRpcService().execute( () -> { final Collection<JobID> jobIds; ...

When the test adds something to the job graph store, it could potentially be submitted by recoverJobs, which is something I need to avoid.

…lbacks Always attempt the job submission if onAddedJobGraph or onRemovedJobGraph are called. The checks in submitJob and removeJob are sufficient.

tillrohrmann · 2017-12-11T08:12:44Z

Changes look good. Will merge this.

Implement SubmittedJobGraphListener interface in Dispatcher Call start() on SubmittedJobGraphStore with Dispatcher as listener. To enable this, the dispatcher must implement the SubmittedJobGraphListener interface. Add simple unit tests for the new methods. Refactor DispatcherTest to remove redundancy. [FLINK-8176][flip6] Make InMemorySubmittedJobGraphStore thread-safe [FLINK-8176][flip6] Add method isStarted() to TestingLeaderElectionService [FLINK-8176][flip6] Return same RunningJobsRegistry instance from TestingHighAvailabilityServices [FLINK-8176][flip6] Fix race conditions in Dispatcher and DispatcherTest Check if jobManagerRunner exists before submitting job. Replace JobManagerRunner mock used in tests with real instance. Do not run job graph recovery in actor main thread when job graph is recovered from SubmittedJobGraphListener#onAddedJobGraph(JobID). [FLINK-8176][flip6] Rename variables in DispatcherTest [FLINK-8176][flip6] Remove injectMocks in DispatcherTest [FLINK-8176][flip6] Update Dispatcher's SubmittedJobGraphListener callbacks Always attempt the job submission if onAddedJobGraph or onRemovedJobGraph are called. The checks in submitJob and removeJob are sufficient. This closes apache#5107.

GJL commented Nov 30, 2017

View reviewed changes

GJL commented Dec 1, 2017

View reviewed changes

GJL closed this Dec 1, 2017

tillrohrmann reviewed Dec 1, 2017

View reviewed changes

GJL reopened this Dec 4, 2017

GJL commented Dec 4, 2017

View reviewed changes

GJL force-pushed the FLINK-8176 branch from 7a04cbe to 092cfb8 Compare December 5, 2017 08:29

GJL added 8 commits December 5, 2017 13:48

[hotfix][tests] Extract SubmittedJobGraphStore implementation from Jo…

0760ae6

…bManagerHARecoveryTest

[FLINK-8176][flip6] Make InMemorySubmittedJobGraphStore thread-safe

461dbe2

[hotfix][Javadoc] Make first sentence in JobSubmissionException Javad…

75d1bce

…oc end with period

[FLINK-8176][flip6] Add method isStarted() to TestingLeaderElectionSe…

fd1a8ab

…rvice

[FLINK-8176][flip6] Return same RunningJobsRegistry instance from Tes…

8da1491

…tingHighAvailabilityServices

[FLINK-8176][flip6] Rename variables in DispatcherTest

3f5e25f

GJL force-pushed the FLINK-8176 branch from 092cfb8 to 3f5e25f Compare December 5, 2017 12:54

tillrohrmann approved these changes Dec 6, 2017

View reviewed changes

GJL added 2 commits December 7, 2017 12:18

[FLINK-8176][flip6] Remove injectMocks in DispatcherTest

7cc0767

[FLINK-8176][flip6] Update Dispatcher's SubmittedJobGraphListener cal…

6170a93

…lbacks Always attempt the job submission if onAddedJobGraph or onRemovedJobGraph are called. The checks in submitJob and removeJob are sufficient.

GJL force-pushed the FLINK-8176 branch from 0864b3a to 6170a93 Compare December 7, 2017 11:18

asfgit closed this in 7ddb674 Dec 13, 2017

rmetzger added component=Deployment/YARN component=Runtime/Coordination labels Mar 18, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-8176][flip6] Start SubmittedJobGraphStore in Dispatcher #5107

[FLINK-8176][flip6] Start SubmittedJobGraphStore in Dispatcher #5107

GJL commented Nov 30, 2017

GJL Nov 30, 2017

GJL Dec 1, 2017

tillrohrmann Dec 1, 2017

tillrohrmann left a comment

tillrohrmann Dec 1, 2017

GJL Dec 4, 2017

tillrohrmann Dec 1, 2017

GJL Dec 4, 2017

tillrohrmann left a comment

tillrohrmann Dec 6, 2017

GJL Dec 6, 2017

tillrohrmann Dec 6, 2017

GJL Dec 6, 2017

tillrohrmann Dec 6, 2017

GJL Dec 6, 2017

tillrohrmann Dec 6, 2017

GJL Dec 6, 2017

GJL Dec 6, 2017 •

edited

tillrohrmann Dec 6, 2017

GJL Dec 6, 2017

tillrohrmann commented Dec 11, 2017

[FLINK-8176][flip6] Start SubmittedJobGraphStore in Dispatcher #5107

[FLINK-8176][flip6] Start SubmittedJobGraphStore in Dispatcher #5107

Conversation

GJL commented Nov 30, 2017

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tillrohrmann left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tillrohrmann left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GJL Dec 6, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tillrohrmann commented Dec 11, 2017

GJL Dec 6, 2017 •

edited