[FLINK-21401] Consolidate JobGraph generation to properly instantiate streaming and batch JobGraphs #15093

tillrohrmann · 2021-03-05T08:55:01Z

This PR consolidates the JobGraph generation into as few places as possible. Moreover it introduces explicit factory methods for streaming and batch JobGraphs (JobType.STREAMING and JobType.BATCH) in order to better classify which tests can be run with the AdaptiveScheduler and which strictly require the DefaultScheduler.

This PR also introduces a JobGraphBuilder which can be used to make the JobGraph an immutable data structure. This builder is then used for the more complex JobGraph creations for tests.

Last but not least, this PR adapts some failing tests (mainly DispatcherTest) to also work with the AdaptiveScheduler.

flinkbot · 2021-03-05T09:05:42Z

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.

Automated Checks

Last check on commit ff8af80 (Fri Mar 05 09:05:42 UTC 2021)

Warnings:

1 pom.xml files were touched: Check for build and licensing issues.
No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

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.

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

flinkbot · 2021-03-05T09:24:32Z

CI report:

364a8ef Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

rmetzger

Thanks a lot for all the cleanups! Except for one minor oversight, I couldn't find any major issues in the change.

+1 to merge once CI passed.

rmetzger · 2021-03-05T11:37:49Z

...e/src/test/java/org/apache/flink/runtime/entrypoint/component/FileJobGraphRetrieverTest.java

+                JobGraphBuilder.newStreamingJobGraphBuilder()
+                        .addJobVertices(Arrays.asList(source, target))
+                        .addClasspaths(Collections.singletonList(jarFileInJobGraph.toUri().toURL()))
+                        .build();

        jobGraph.setClasspaths(Collections.singletonList(jarFileInJobGraph.toUri().toURL()));


Suggested change

jobGraph.setClasspaths(Collections.singletonList(jarFileInJobGraph.toUri().toURL()));

Good catch. Will remove it.

rmetzger · 2021-03-05T12:33:44Z

flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java

@@ -260,100 +245,6 @@ public static void teardownClass() {
        }
    }

-    @Test
-    public void testDeclineCheckpointInvocationWithUserException() throws Exception {


found https://github.com/apache/flink/pull/14662/files#r585721398 ;)

I should have added a reference to this discussion. Thanks for digging it up.

rmetzger · 2021-03-05T13:25:31Z

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

            fail("Previous statement should have failed");
        } catch (ExecutionException t) {
            assertTrue(t.getCause() instanceof UnavailableDispatcherOperationException);
        }

        // submission has succeeded, let the initialization finish.
-        blockingJobGraph.f1.unblock();
+        latch.trigger();


This is much nicer than the Tuple2 with the vertex! Thanks for the cleanup.

Yeah, Tuple2 is actually quite ugly. We should try to never use it if possible.

rmetzger · 2021-03-05T13:30:18Z

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

+                        .setJobId(jobId)
+                        .addJobVertex(blockingJobVertex)
+                        .build(),
+                blockingJobVertex);
    }

    private static class FailingJobVertex extends JobVertex {


Thanks a lot for the various cleanups! I took some notes.

I also added a couple of tests which relied on scheduler details. I think before we did not have the necessity to separate tests thoroughly.

The AdaptiveScheduler needs to make the JobVertices serializable and let them use static fields to communicate with the test because the JobGraph is copied. This closes apache#15093.

tillrohrmann · 2021-03-05T14:51:18Z

Thanks for the review @rmetzger. I've addressed your comments. I will merge this PR once AZP gives green light.

The AdaptiveScheduler needs to make the JobVertices serializable and let them use static fields to communicate with the test because the JobGraph is copied. This closes apache#15093.

…rFactoryTest to use streaming jobs

…hUserException to not need JobMaster instance

…til factories

…nstructor

…endent of underlying JobMaster implementation This commit makes the DispatcherTest.testErrorDuringInitialization independent of the underlying JobMaster implementation by splitting it into DispatcherTest.testJobManagerRunnerInitializationFailureFailsJob and JobManagerRunnerImplTest.testJobMasterCreationFailureCompletesJobManagerRunnerWithInitializationError which tests in two steps what DispatcherTest.testErrorDuringInitialization tested by relying on the DefaultScheduler to eagerly create an ExecutionGraph and to fail.

… not depend on underlying Scheduler implementation By using the BlockingJobManagerRunnerFactory it is possible to decouple the DispatcherTest.testInvalidCallDuringInitialization from the underlying Scheduler implementation which was used for blocking the creation of the JobManagerRunner before.

…ndependent of Scheduler implementation By letting the DispatcherTest.testWaitingForJobMasterLeadership use the TestingJobManagerRunnerFactory we can abstract this test from the implementation details of the Scheduler and when it shows which JobStatus.

…dToExecutionGraph independent of Scheduler implementation Change Dispatcher.testInitializationTimestampForwardedToExecutionGraph to testInitializationTimestampForwardedToJobManagerRunner which only tests that the initialization timestamp is forwarded to the JobManagerRunner. Additionally, this commit adds AdaptiveSchedulerTest.testExecutionGraphGenerationSetsInitializationTimestamp, AdaptiveSchedulerTest.testInitializationTimestampForwarding and DefaultSchedulerTest.testCorrectSettingOfInitializationTimestamp to test the timestamp forwarding for the different Scheduler implementations.

…BeStarted independent of Scheduler implementation This commit makes the DispatcherTest.testFatalErrorIfRecoveredJobsCannotBeStarted independent of the scheduler implementation by using the TestingJobManagerRunnerFactory and letting the TestingJobManagerRunner complete with an initialization error.

…ndent of underlying scheduler Make DispatcherTest.testNonBlockingJobSubmission independent of underlying scheduler by using the BlockingJobManagerRunnerFactory and blocking the creation of the JobManagerRunner.

…ot needed The leader elections are only needed where we instantiate a proper JobManagerRunnerImpl which needs leader election.

The AdaptiveScheduler needs to make the JobVertices serializable and let them use static fields to communicate with the test because the JobGraph is copied. This closes apache#15093.

…nough slots for test jobs

…or the adaptive scheduler The adaptive scheduler does not support stop with savepoint yet. Therefore, this test cannot work.

…otCreation and .testStopWithSavepointFailingAfterSnapshotCreation for the AdaptiveScheduler The AdaptiveScheduler does not support stopping jobs with savepoint yet.

…ncreasing resource timeout to 30s for the AdaptiveScheduler The test seems to take some time to start the required TaskManagers. This can sometimes lead to resource timeouts in the WaitingForResources state if it takes too long. This commit fixes this problem by increasing the resource timeout to 30s.

The AdaptiveScheduler needs to make the JobVertices serializable and let them use static fields to communicate with the test because the JobGraph is copied. This closes apache#15093.

rmetzger added review=description? component=Runtime/Coordination component=Tests labels Mar 5, 2021

rmetzger approved these changes Mar 5, 2021

View reviewed changes

tillrohrmann force-pushed the FLINK-21401 branch from 80919f4 to c1139e5 Compare March 5, 2021 14:50

tillrohrmann force-pushed the FLINK-21401 branch from c1139e5 to c285494 Compare March 5, 2021 18:36

tillrohrmann added 19 commits March 5, 2021 19:37

[FLINK-21401] Introduce JobGraphTestUtils and rework PerJobMiniCluste…

dc051fe

…rFactoryTest to use streaming jobs

[FLINK-21401] Consolidate different JobGraphTestUtils into one class

6f4fce9

[FLINK-21401] Use JobGraphTestUtils in JobGraphGeneratorTest

a27515b

[FLINK-21401] Use JobGraphTestUtils in ClientUtilsTest

9b36632

[FLINK-21401] Use JobGraphTestUtils in MiniDispatcherTest

ed641d7

[FLINK-21401] Replace manual JobGraph creation with factory methods

963bb75

[hotfix][tests] Refactor JobMaster.testDeclineCheckpointInvocationWit…

8509dab

…hUserException to not need JobMaster instance

[FLINK-21401] Replace explicit JobGraph generation with JobGraphTestU…

88c2d10

…til factories

[hotfix] Add @nullable annotation to JobGraph constructors

77b327c

[FLINK-21401] Add JobGraphBuilder and adapt call sites of JobGraph co…

e000723

…nstructor

[FLINK-21401] Replace JobGraph constructors with JobGraphBuilder

387bb06

[hotfix] Let ApplicationDispatcherBootstrapTest extend TestLogger

f6655b1

[FLINK-21401] Make DispatcherTest.testNonBlockingJobSubmission indepe…

08a0c71

…ndent of underlying scheduler Make DispatcherTest.testNonBlockingJobSubmission independent of underlying scheduler by using the BlockingJobManagerRunnerFactory and blocking the creation of the JobManagerRunner.

[hotfix] Remove leader elections from DispatcherTest where they are n…

2994538

…ot needed The leader elections are only needed where we instantiate a proper JobManagerRunnerImpl which needs leader election.

tillrohrmann added 6 commits March 5, 2021 19:37

[FLINK-21401] Make MiniClusterITCase work with AdaptiveScheduler

3926b5e

The AdaptiveScheduler needs to make the JobVertices serializable and let them use static fields to communicate with the test because the JobGraph is copied. This closes apache#15093.

[hotfix] Speed up MiniClusterITCase for AdaptiveScheduler by giving e…

0341958

…nough slots for test jobs

[hotfix] Let HistoryServerStaticFileServerHandlerTest extend TestLogger

31e7477

[hotfix] Ignore PerJobMiniClusterFactoryTest.testJobClientSavepoint f…

07eb7a9

…or the adaptive scheduler The adaptive scheduler does not support stop with savepoint yet. Therefore, this test cannot work.

[hotfix] Disable SavepointITCase.testStopWithSavepointFailingInSnapsh…

ef6dc2c

…otCreation and .testStopWithSavepointFailingAfterSnapshotCreation for the AdaptiveScheduler The AdaptiveScheduler does not support stopping jobs with savepoint yet.

tillrohrmann force-pushed the FLINK-21401 branch from c285494 to 364a8ef Compare March 5, 2021 18:38

tillrohrmann closed this in 3926b5e Mar 6, 2021

tillrohrmann merged commit 364a8ef into apache:master Mar 6, 2021

tillrohrmann deleted the FLINK-21401 branch March 6, 2021 12:25

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-21401] Consolidate JobGraph generation to properly instantiate streaming and batch JobGraphs #15093

[FLINK-21401] Consolidate JobGraph generation to properly instantiate streaming and batch JobGraphs #15093

tillrohrmann commented Mar 5, 2021

flinkbot commented Mar 5, 2021

flinkbot commented Mar 5, 2021 •

edited

rmetzger left a comment

rmetzger Mar 5, 2021

tillrohrmann Mar 5, 2021

rmetzger Mar 5, 2021

tillrohrmann Mar 5, 2021

rmetzger Mar 5, 2021

tillrohrmann Mar 5, 2021

rmetzger Mar 5, 2021

tillrohrmann Mar 5, 2021

tillrohrmann commented Mar 5, 2021

[FLINK-21401] Consolidate JobGraph generation to properly instantiate streaming and batch JobGraphs #15093

[FLINK-21401] Consolidate JobGraph generation to properly instantiate streaming and batch JobGraphs #15093

Conversation

tillrohrmann commented Mar 5, 2021

flinkbot commented Mar 5, 2021

Automated Checks

Review Progress

flinkbot commented Mar 5, 2021 • edited

CI report:

rmetzger 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

tillrohrmann commented Mar 5, 2021

flinkbot commented Mar 5, 2021 •

edited