[FLINK-33212][runtime] add job status changed listener for lineage #24754

HuangZhenQiu · 2024-05-01T17:09:00Z

What is the purpose of the change

This PR is to implement the job status changed listener for lineage in FLIP-314. This PR use class loader to load listeners in pipeline executors, including LocalExecutor, EmbeddedExecutor for application mode and AbstractSessionClusterExecutor for session cluster. When job graph submission is successfully, the JobCreatedEvent with lineage info will be published to listeners. During the runtime, job status change info will also be published to listeners.

Brief change log

Add interfaces for job status change listener
Add Events and configs for job status listener.
Add logic pipeline executors for notifying the job created event for local, application mode, and session cluster.

Verifying this change

This change added tests and can be verified as follows:

The end to end test is covered by JobStatusListenerITCase

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

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

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (not documented)

HuangZhenQiu · 2024-05-01T17:10:13Z

...c/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java

@@ -80,7 +80,7 @@ public CompletableFuture<JobClient> execute(
                    clusterDescriptor.retrieve(clusterID);
            ClusterClient<ClusterID> clusterClient = clusterClientProvider.getClusterClient();
            return clusterClient
-                    .submitJob(jobGraph)
+                    .submitJob(jobGraph, pipeline)


@PatrickRen How should we send lineagegraph into the cluster client without changing API?

I think that you are saying that the original submitJob with one parameter still works. As well as the new method.

Yes, but there is a concern about the size of lineage info if put into job graph.

flinkbot · 2024-05-01T17:12:12Z

CI report:

c771823 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java

flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java

flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java

gyfora · 2024-05-08T16:10:29Z

flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java

@@ -74,7 +75,7 @@ public Configuration getFlinkConfiguration() {
    }

    @Override
-    public CompletableFuture<JobID> submitJob(@Nonnull JobGraph jobGraph) {
+    public CompletableFuture<JobID> submitJob(@Nonnull JobGraph jobGraph, Pipeline pipeline) {


We discussed this with @HuangZhenQiu offline and we should remove this api change and the lineage logic will be applied on the future instead

gyfora · 2024-05-13T10:19:25Z

...c/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java

+    private final ExecutorService executorService =
+            Executors.newFixedThreadPool(
+                    4, new ExecutorThreadFactory("Flink-SessionClusterExecutor-IO"));


Why do we need 4 threads? Do we expect concurrent calls here?

...c/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java

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

docs/content/docs/deployment/advanced/job_status_listener.md

HuangZhenQiu · 2024-05-20T04:30:05Z

@gyfora @JingGe
Thanks for your review. I have partially resolved your comments. The lineage graph info is set into StreamGraph in this correlated PR https://github.com/apache/flink/pull/24618/files. I would also like to get more suggestion about how to make PRs more self-contained.

HuangZhenQiu · 2024-05-21T01:59:35Z

@flinkbot run azure

JingGe · 2024-05-21T15:48:38Z

@flinkbot run azure

davidradl · 2024-05-22T12:12:02Z

...src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutor.java

+                        jobID -> jobClientCreator.getJobClient(actualJobId, userCodeClassloader))
+                .whenCompleteAsync(
+                        (jobClient, throwable) -> {
+                            if (throwable == null) {


Should we do something if there is a throwable . Maybe notify to the job status changed listeners that there was an error and log the error. If there is a good reason to swallow the throwable here, then a comment explaining would be good.

davidradl · 2024-05-22T12:12:51Z

...c/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java

-                    .whenCompleteAsync((ignored1, ignored2) -> clusterClient.close());
+                    .whenCompleteAsync(
+                            (jobClient, throwable) -> {
+                                if (throwable == null) {


same comment as above - what happens if there is a throwable

davidradl · 2024-05-22T12:16:15Z

flink-clients/src/main/java/org/apache/flink/client/deployment/executors/LocalExecutor.java

+                .submitJob(jobGraph, userCodeClassloader)
+                .whenComplete(
+                        (ignored, throwable) -> {
+                            if (throwable == null) {


same comment as above

HuangZhenQiu · 2024-05-22T17:36:46Z

@davidradl
The throwable in executors are caught already in Execution environment. If there is a better idea to provide extra info for customers, I am glad to adopt.

HuangZhenQiu commented May 1, 2024

View reviewed changes

HuangZhenQiu force-pushed the FLINK-33212-lineage-listener branch from 6b5b888 to 18d5d7a Compare May 1, 2024 17:12

flinkbot added the component=Runtime/REST label May 1, 2024

HuangZhenQiu force-pushed the FLINK-33212-lineage-listener branch from 18d5d7a to d904d63 Compare May 1, 2024 20:26

davidradl reviewed May 3, 2024

View reviewed changes

flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java Outdated Show resolved Hide resolved

davidradl reviewed May 3, 2024

View reviewed changes

flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java Outdated Show resolved Hide resolved

davidradl reviewed May 3, 2024

View reviewed changes

flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java Outdated Show resolved Hide resolved

davidradl reviewed May 3, 2024

View reviewed changes

flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java Outdated Show resolved Hide resolved

HuangZhenQiu force-pushed the FLINK-33212-lineage-listener branch from d904d63 to 555ba41 Compare May 7, 2024 02:25

gyfora reviewed May 8, 2024

View reviewed changes

HuangZhenQiu force-pushed the FLINK-33212-lineage-listener branch 2 times, most recently from bdaf645 to 5008234 Compare May 8, 2024 21:57

gyfora reviewed May 13, 2024

View reviewed changes

JingGe reviewed May 14, 2024

View reviewed changes

docs/content/docs/deployment/advanced/job_status_listener.md Show resolved Hide resolved

HuangZhenQiu force-pushed the FLINK-33212-lineage-listener branch from 5008234 to ca78b3c Compare May 20, 2024 04:19

HuangZhenQiu force-pushed the FLINK-33212-lineage-listener branch from ca78b3c to 0f2960d Compare May 20, 2024 22:00

HuangZhenQiu force-pushed the FLINK-33212-lineage-listener branch from 0f2960d to 6cfbd92 Compare May 21, 2024 20:34

[FLINK-33212][core] add job status changed listener for lineage

c771823

HuangZhenQiu force-pushed the FLINK-33212-lineage-listener branch from 6cfbd92 to c771823 Compare May 22, 2024 02:57

davidradl reviewed May 22, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-33212][runtime] add job status changed listener for lineage #24754

[FLINK-33212][runtime] add job status changed listener for lineage #24754

HuangZhenQiu commented May 1, 2024 •

edited

HuangZhenQiu May 1, 2024

davidradl May 3, 2024

HuangZhenQiu May 7, 2024

flinkbot commented May 1, 2024 •

edited

gyfora May 8, 2024

gyfora May 13, 2024

HuangZhenQiu commented May 20, 2024

HuangZhenQiu commented May 21, 2024

JingGe commented May 21, 2024

davidradl May 22, 2024 •

edited

davidradl May 22, 2024

davidradl May 22, 2024

HuangZhenQiu commented May 22, 2024

[FLINK-33212][runtime] add job status changed listener for lineage #24754

Are you sure you want to change the base?

[FLINK-33212][runtime] add job status changed listener for lineage #24754

Conversation

HuangZhenQiu commented May 1, 2024 • edited

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

HuangZhenQiu May 1, 2024

Choose a reason for hiding this comment

davidradl May 3, 2024

Choose a reason for hiding this comment

HuangZhenQiu May 7, 2024

Choose a reason for hiding this comment

flinkbot commented May 1, 2024 • edited

CI report:

gyfora May 8, 2024

Choose a reason for hiding this comment

gyfora May 13, 2024

Choose a reason for hiding this comment

HuangZhenQiu commented May 20, 2024

HuangZhenQiu commented May 21, 2024

JingGe commented May 21, 2024

davidradl May 22, 2024 • edited

Choose a reason for hiding this comment

davidradl May 22, 2024

Choose a reason for hiding this comment

davidradl May 22, 2024

Choose a reason for hiding this comment

HuangZhenQiu commented May 22, 2024

HuangZhenQiu commented May 1, 2024 •

edited

flinkbot commented May 1, 2024 •

edited

davidradl May 22, 2024 •

edited