This is an automated email from the ASF dual-hosted git repository.
zjureel pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 99cb5c7addc [FLINK-33212][core] add job status changed listener for
lineage (#24754)
99cb5c7addc is described below
commit 99cb5c7addc932aba517ea4481934ffe2b162052
Author: Peter Huang <huangzhenqiu0...@gmail.com>
AuthorDate: Mon Jul 1 22:16:16 2024 -0700
[FLINK-33212][core] add job status changed listener for lineage (#24754)
---
.../deployment/advanced/job_status_listener.md | 81 +++++++
.../generated/deployment_configuration.html | 6 +
.../application/executors/EmbeddedExecutor.java | 33 ++-
.../executors/EmbeddedExecutorFactory.java | 1 +
.../executors/WebSubmissionExecutorFactory.java | 1 +
.../executors/AbstractSessionClusterExecutor.java | 37 ++-
.../client/deployment/executors/LocalExecutor.java | 31 ++-
.../executors/PipelineExecutorUtils.java | 41 ++++
.../deployment/executors/RemoteExecutor.java | 5 +-
.../executors/RemoteExecutorFactory.java | 2 +-
.../flink/configuration/DeploymentOptions.java | 8 +
.../execution/DefaultJobExecutionStatusEvent.java | 74 ++++++
.../core/execution/JobExecutionStatusEvent.java | 28 ++-
.../core/execution/JobStatusChangedEvent.java | 21 +-
.../core/execution/JobStatusChangedListener.java | 24 +-
.../execution/JobStatusChangedListenerFactory.java | 54 +++++
.../execution/JobStatusChangedListenerUtils.java | 79 ++++++
.../KubernetesSessionClusterExecutor.java | 5 +-
.../KubernetesSessionClusterExecutorFactory.java | 2 +-
.../executiongraph/DefaultExecutionGraph.java | 22 +-
.../DefaultExecutionGraphBuilder.java | 9 +-
.../flink/streaming/api/graph/StreamGraph.java | 6 +
.../runtime/execution/DefaultJobCreatedEvent.java | 64 +++++
.../runtime/execution/JobCreatedEvent.java | 25 +-
.../MiniClusterPipelineExecutorServiceLoader.java | 27 ++-
.../execution/JobStatusChangedListenerITCase.java | 266 +++++++++++++++++++++
.../yarn/executors/YarnSessionClusterExecutor.java | 5 +-
.../YarnSessionClusterExecutorFactory.java | 2 +-
28 files changed, 888 insertions(+), 71 deletions(-)
diff --git a/docs/content/docs/deployment/advanced/job_status_listener.md
b/docs/content/docs/deployment/advanced/job_status_listener.md
new file mode 100644
index 00000000000..723cc862594
--- /dev/null
+++ b/docs/content/docs/deployment/advanced/job_status_listener.md
@@ -0,0 +1,81 @@
+
+---
+title: "Job Status Changed Listener"
+nav-title: job-status-listener
+nav-parent_id: advanced
+nav-pos: 3
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+## Job status changed listener
+Flink provides a pluggable interface for users to register their custom logic
for handling with the job status changes in which lineage info about
source/sink is provided.
+This enables users to implement their own flink lineage reporter to send
lineage info to third party data lineage systems for example Datahub and
Openlineage.
+
+The job status changed listeners are triggered every time status change
happened for the application. The data lineage info is included in the
JobCreatedEvent.
+
+### Implement a plugin for your custom enricher
+
+To implement a custom JobStatusChangedListener plugin, you need to:
+
+- Add your own JobStatusChangedListener by implementing the {{< gh_link
file="/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListener.java"
name="JobStatusChangedListener" >}} interface.
+
+- Add your own JobStatusChangedListenerFactory by implementing the {{< gh_link
file="/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListenerFactory.java"
name="JobStatusChangedListenerFactory" >}} interface.
+
+- Add a service entry. Create a file
`META-INF/services/org.apache.flink.core.execution.JobStatusChangedListenerFactory`
which contains the class name of your job status changed listener factory
class (see [Java Service
Loader](https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/ServiceLoader.html)
docs for more details).
+
+
+Then, create a jar which includes your `JobStatusChangedListener`,
`JobStatusChangedListenerFactory`, `META-INF/services/` and all external
dependencies.
+Make a directory in `plugins/` of your Flink distribution with an arbitrary
name, e.g. "job-status-changed-listener", and put the jar into this directory.
+See [Flink Plugin]({{< ref "docs/deployment/filesystems/plugins" >}}) for more
details.
+
+JobStatusChangedListenerFactory example:
+
+``` java
+package org.apache.flink.test.execution;
+
+public static class TestingJobStatusChangedListenerFactory
+ implements JobStatusChangedListenerFactory {
+
+ @Override
+ public JobStatusChangedListener createListener(Context context) {
+ return new TestingJobStatusChangedListener();
+ }
+}
+```
+
+JobStatusChangedListener example:
+
+``` java
+package org.apache.flink.test.execution;
+
+private static class TestingJobStatusChangedListener implements
JobStatusChangedListener {
+
+ @Override
+ public void onEvent(JobStatusChangedEvent event) {
+ statusChangedEvents.add(event);
+ }
+}
+```
+
+### Configuration
+
+Flink components loads JobStatusChangedListener plugins at startup. To make
sure your JobStatusChangedListeners are loaded all class names should be
defined as part of [execution.job-status-changed-listeners]({{< ref
"docs/deployment/config#execution.job-status-changed-listeners" >}}).
+ If this configuration is empty, NO enrichers will be started. Example:
+```
+ execution.job-status-changed-listeners =
org.apache.flink.test.execution.TestingJobStatusChangedListenerFactory
+```
diff --git a/docs/layouts/shortcodes/generated/deployment_configuration.html
b/docs/layouts/shortcodes/generated/deployment_configuration.html
index 07daa602a07..f3e08ecd696 100644
--- a/docs/layouts/shortcodes/generated/deployment_configuration.html
+++ b/docs/layouts/shortcodes/generated/deployment_configuration.html
@@ -20,6 +20,12 @@
<td>List<String></td>
<td>Custom JobListeners to be registered with the execution
environment. The registered listeners cannot have constructors with
arguments.</td>
</tr>
+ <tr>
+ <td><h5>execution.job-status-changed-listeners</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>List<String></td>
+ <td>When job is created or its status is changed, Flink will
generate job event and notify job status changed listener.</td>
+ </tr>
<tr>
<td><h5>execution.program-config.enabled</h5></td>
<td style="word-wrap: break-word;">true</td>
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutor.java
b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutor.java
index 7094c1cac13..408bc69ad31 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutor.java
+++
b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutor.java
@@ -27,12 +27,15 @@ import
org.apache.flink.client.deployment.executors.PipelineExecutorUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.PipelineOptionsInternal;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobStatusChangedListener;
+import org.apache.flink.core.execution.JobStatusChangedListenerUtils;
import org.apache.flink.core.execution.PipelineExecutor;
import org.apache.flink.runtime.blob.BlobClient;
import org.apache.flink.runtime.client.ClientUtils;
import org.apache.flink.runtime.dispatcher.DispatcherGateway;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.function.FunctionUtils;
import org.slf4j.Logger;
@@ -41,9 +44,12 @@ import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
import java.net.MalformedURLException;
import java.util.Collection;
+import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -57,6 +63,10 @@ public class EmbeddedExecutor implements PipelineExecutor {
private static final Logger LOG =
LoggerFactory.getLogger(EmbeddedExecutor.class);
+ private final ExecutorService executorService =
+ Executors.newFixedThreadPool(
+ 1, new
ExecutorThreadFactory("Flink-EmbeddedClusterExecutor-IO"));
+
public static final String NAME = "embedded";
private final Collection<JobID> submittedJobIds;
@@ -65,6 +75,8 @@ public class EmbeddedExecutor implements PipelineExecutor {
private final EmbeddedJobClientCreator jobClientCreator;
+ private final List<JobStatusChangedListener> jobStatusChangedListeners;
+
/**
* Creates a {@link EmbeddedExecutor}.
*
@@ -73,14 +85,22 @@ public class EmbeddedExecutor implements PipelineExecutor {
* caller.
* @param dispatcherGateway the dispatcher of the cluster which is going
to be used to submit
* jobs.
+ * @param configuration the flink application configuration
+ * @param jobClientCreator the job client creator
*/
public EmbeddedExecutor(
final Collection<JobID> submittedJobIds,
final DispatcherGateway dispatcherGateway,
+ final Configuration configuration,
final EmbeddedJobClientCreator jobClientCreator) {
this.submittedJobIds = checkNotNull(submittedJobIds);
this.dispatcherGateway = checkNotNull(dispatcherGateway);
this.jobClientCreator = checkNotNull(jobClientCreator);
+ this.jobStatusChangedListeners =
+ JobStatusChangedListenerUtils.createJobStatusChangedListeners(
+ Thread.currentThread().getContextClassLoader(),
+ configuration,
+ executorService);
}
@Override
@@ -153,7 +173,18 @@ public class EmbeddedExecutor implements PipelineExecutor {
return jobId;
}))
.thenApplyAsync(
- jobID -> jobClientCreator.getJobClient(actualJobId,
userCodeClassloader));
+ jobID -> jobClientCreator.getJobClient(actualJobId,
userCodeClassloader))
+ .whenCompleteAsync(
+ (jobClient, throwable) -> {
+ if (throwable == null) {
+ PipelineExecutorUtils.notifyJobStatusListeners(
+ pipeline, jobGraph,
jobStatusChangedListeners);
+ } else {
+ LOG.error(
+ "Failed to submit job graph to
application cluster",
+ throwable);
+ }
+ });
}
private static CompletableFuture<JobID> submitJob(
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutorFactory.java
b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutorFactory.java
index 0810c1e1c06..790fffb0a1f 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutorFactory.java
+++
b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutorFactory.java
@@ -79,6 +79,7 @@ public class EmbeddedExecutorFactory implements
PipelineExecutorFactory {
return new EmbeddedExecutor(
submittedJobIds,
dispatcherGateway,
+ configuration,
(jobId, userCodeClassloader) -> {
final Time timeout =
Time.milliseconds(
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/WebSubmissionExecutorFactory.java
b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/WebSubmissionExecutorFactory.java
index f087b5f30db..16e3bf05813 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/WebSubmissionExecutorFactory.java
+++
b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/executors/WebSubmissionExecutorFactory.java
@@ -75,6 +75,7 @@ public class WebSubmissionExecutorFactory implements
PipelineExecutorFactory {
return new EmbeddedExecutor(
submittedJobIds,
dispatcherGateway,
+ configuration,
(jobId, userCodeClassloader) -> new
WebSubmissionJobClient(jobId));
}
}
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java
b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java
index 9397e506ccf..7c22131a4d5 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java
+++
b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java
@@ -29,16 +29,25 @@ import
org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.CacheSupportedPipelineExecutor;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobStatusChangedListener;
+import org.apache.flink.core.execution.JobStatusChangedListenerUtils;
import org.apache.flink.core.execution.PipelineExecutor;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.util.AbstractID;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.function.FunctionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nonnull;
+import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
@@ -55,11 +64,24 @@ import static
org.apache.flink.util.Preconditions.checkState;
public class AbstractSessionClusterExecutor<
ClusterID, ClientFactory extends
ClusterClientFactory<ClusterID>>
implements CacheSupportedPipelineExecutor {
+ private static final Logger LOG =
LoggerFactory.getLogger(AbstractSessionClusterExecutor.class);
+ private final ExecutorService executorService =
+ Executors.newFixedThreadPool(
+ 1, new
ExecutorThreadFactory("Flink-SessionClusterExecutor-IO"));
private final ClientFactory clusterClientFactory;
+ private final Configuration configuration;
+ private final List<JobStatusChangedListener> jobStatusChangedListeners;
- public AbstractSessionClusterExecutor(@Nonnull final ClientFactory
clusterClientFactory) {
+ public AbstractSessionClusterExecutor(
+ @Nonnull final ClientFactory clusterClientFactory, Configuration
configuration) {
this.clusterClientFactory = checkNotNull(clusterClientFactory);
+ this.configuration = configuration;
+ this.jobStatusChangedListeners =
+ JobStatusChangedListenerUtils.createJobStatusChangedListeners(
+ Thread.currentThread().getContextClassLoader(),
+ configuration,
+ executorService);
}
@Override
@@ -97,7 +119,18 @@ public class AbstractSessionClusterExecutor<
clusterClientProvider,
jobID,
userCodeClassloader))
- .whenCompleteAsync((ignored1, ignored2) ->
clusterClient.close());
+ .whenCompleteAsync(
+ (jobClient, throwable) -> {
+ if (throwable == null) {
+
PipelineExecutorUtils.notifyJobStatusListeners(
+ pipeline, jobGraph,
jobStatusChangedListeners);
+ } else {
+ LOG.error(
+ "Failed to submit job graph to
remote session cluster.",
+ throwable);
+ }
+ clusterClient.close();
+ });
}
}
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/LocalExecutor.java
b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/LocalExecutor.java
index b8b60fe9adf..27e1043a785 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/LocalExecutor.java
+++
b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/LocalExecutor.java
@@ -26,13 +26,22 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobStatusChangedListener;
+import org.apache.flink.core.execution.JobStatusChangedListenerUtils;
import org.apache.flink.core.execution.PipelineExecutor;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.net.MalformedURLException;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.function.Function;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -41,11 +50,15 @@ import static
org.apache.flink.util.Preconditions.checkState;
/** An {@link PipelineExecutor} for executing a {@link Pipeline} locally. */
@Internal
public class LocalExecutor implements PipelineExecutor {
+ private static final Logger LOG =
LoggerFactory.getLogger(LocalExecutor.class);
+ private final ExecutorService executorService =
+ Executors.newFixedThreadPool(1, new
ExecutorThreadFactory("Flink-LocalExecutor-IO"));
public static final String NAME = "local";
private final Configuration configuration;
private final Function<MiniClusterConfiguration, MiniCluster>
miniClusterFactory;
+ private final List<JobStatusChangedListener> jobStatusChangedListeners;
public static LocalExecutor create(Configuration configuration) {
return new LocalExecutor(configuration, MiniCluster::new);
@@ -62,6 +75,11 @@ public class LocalExecutor implements PipelineExecutor {
Function<MiniClusterConfiguration, MiniCluster>
miniClusterFactory) {
this.configuration = configuration;
this.miniClusterFactory = miniClusterFactory;
+ this.jobStatusChangedListeners =
+ JobStatusChangedListenerUtils.createJobStatusChangedListeners(
+ Thread.currentThread().getContextClassLoader(),
+ configuration,
+ executorService);
}
@Override
@@ -81,7 +99,18 @@ public class LocalExecutor implements PipelineExecutor {
final JobGraph jobGraph = getJobGraph(pipeline, effectiveConfig,
userCodeClassloader);
return PerJobMiniClusterFactory.createWithFactory(effectiveConfig,
miniClusterFactory)
- .submitJob(jobGraph, userCodeClassloader);
+ .submitJob(jobGraph, userCodeClassloader)
+ .whenComplete(
+ (ignored, throwable) -> {
+ if (throwable == null) {
+ PipelineExecutorUtils.notifyJobStatusListeners(
+ pipeline, jobGraph,
jobStatusChangedListeners);
+ } else {
+ LOG.error(
+ "Failed to submit job graph to local
mini cluster.",
+ throwable);
+ }
+ });
}
private JobGraph getJobGraph(
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/PipelineExecutorUtils.java
b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/PipelineExecutorUtils.java
index 2981492a9cd..3617bc01ba5 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/PipelineExecutorUtils.java
+++
b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/PipelineExecutorUtils.java
@@ -19,23 +19,33 @@
package org.apache.flink.client.deployment.executors;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.dag.Pipeline;
import org.apache.flink.client.FlinkPipelineTranslationUtil;
import org.apache.flink.client.cli.ClientOptions;
import org.apache.flink.client.cli.ExecutionConfigAccessor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DeploymentOptions;
+import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.configuration.PipelineOptionsInternal;
+import org.apache.flink.core.execution.JobStatusChangedListener;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.streaming.runtime.execution.DefaultJobCreatedEvent;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
import java.net.MalformedURLException;
+import java.util.List;
import static org.apache.flink.util.Preconditions.checkNotNull;
/** Utility class with method related to job execution. */
public class PipelineExecutorUtils {
+ private static final Logger LOG =
LoggerFactory.getLogger(PipelineExecutorUtils.class);
/**
* Creates the {@link JobGraph} corresponding to the provided {@link
Pipeline}.
@@ -80,4 +90,35 @@ public class PipelineExecutorUtils {
return jobGraph;
}
+
+ /**
+ * Notify the {@link DefaultJobCreatedEvent} to job status changed
listeners.
+ *
+ * @param pipeline the pipeline that contains lineage graph information.
+ * @param jobGraph jobGraph that contains job basic info
+ * @param listeners the list of job status changed listeners
+ */
+ public static void notifyJobStatusListeners(
+ @Nonnull final Pipeline pipeline,
+ @Nonnull final JobGraph jobGraph,
+ List<JobStatusChangedListener> listeners) {
+ RuntimeExecutionMode executionMode =
+
jobGraph.getJobConfiguration().get(ExecutionOptions.RUNTIME_MODE);
+ listeners.forEach(
+ listener -> {
+ try {
+ listener.onEvent(
+ new DefaultJobCreatedEvent(
+ jobGraph.getJobID(),
+ jobGraph.getName(),
+ ((StreamGraph)
pipeline).getLineageGraph(),
+ executionMode));
+ } catch (Throwable e) {
+ LOG.error(
+ "Fail to notify job status changed listener
{}",
+ listener.getClass().getName(),
+ e);
+ }
+ });
+ }
}
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
index 761d3cc9141..cbf66cd5edb 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
+++
b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
@@ -21,6 +21,7 @@ package org.apache.flink.client.deployment.executors;
import org.apache.flink.annotation.Internal;
import org.apache.flink.client.deployment.StandaloneClientFactory;
import org.apache.flink.client.deployment.StandaloneClusterId;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.PipelineExecutor;
/** The {@link PipelineExecutor} to be used when executing a job on an already
running cluster. */
@@ -30,7 +31,7 @@ public class RemoteExecutor
public static final String NAME = "remote";
- public RemoteExecutor() {
- super(new StandaloneClientFactory());
+ public RemoteExecutor(Configuration configuration) {
+ super(new StandaloneClientFactory(), configuration);
}
}
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutorFactory.java
b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutorFactory.java
index 5548156aa0c..bbcb7465e35 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutorFactory.java
+++
b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutorFactory.java
@@ -40,6 +40,6 @@ public class RemoteExecutorFactory implements
PipelineExecutorFactory {
@Override
public PipelineExecutor getExecutor(final Configuration configuration) {
- return new RemoteExecutor();
+ return new RemoteExecutor(configuration);
}
}
diff --git
a/flink-core/src/main/java/org/apache/flink/configuration/DeploymentOptions.java
b/flink-core/src/main/java/org/apache/flink/configuration/DeploymentOptions.java
index f6ca0d8a345..aeb58a59bc5 100644
---
a/flink-core/src/main/java/org/apache/flink/configuration/DeploymentOptions.java
+++
b/flink-core/src/main/java/org/apache/flink/configuration/DeploymentOptions.java
@@ -78,6 +78,14 @@ public class DeploymentOptions {
"Custom JobListeners to be registered with the
execution environment."
+ " The registered listeners cannot have
constructors with arguments.");
+ public static final ConfigOption<List<String>>
JOB_STATUS_CHANGED_LISTENERS =
+ key("execution.job-status-changed-listeners")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ "When job is created or its status is changed,
Flink will generate job event and notify job status changed listener.");
+
public static final ConfigOption<Boolean> SHUTDOWN_ON_APPLICATION_FINISH =
ConfigOptions.key("execution.shutdown-on-application-finish")
.booleanType()
diff --git
a/flink-core/src/main/java/org/apache/flink/core/execution/DefaultJobExecutionStatusEvent.java
b/flink-core/src/main/java/org/apache/flink/core/execution/DefaultJobExecutionStatusEvent.java
new file mode 100644
index 00000000000..d8cb2da4568
--- /dev/null
+++
b/flink-core/src/main/java/org/apache/flink/core/execution/DefaultJobExecutionStatusEvent.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+
+import javax.annotation.Nullable;
+
+/** Default implementation for {@link JobExecutionStatusEvent}. */
+@Internal
+public class DefaultJobExecutionStatusEvent implements JobExecutionStatusEvent
{
+ private final JobID jobId;
+ private final String jobName;
+ private final JobStatus oldStatus;
+ private final JobStatus newStatus;
+ @Nullable private final Throwable cause;
+
+ public DefaultJobExecutionStatusEvent(
+ JobID jobId,
+ String jobName,
+ JobStatus oldStatus,
+ JobStatus newStatus,
+ @Nullable Throwable cause) {
+ this.jobId = jobId;
+ this.jobName = jobName;
+ this.oldStatus = oldStatus;
+ this.newStatus = newStatus;
+ this.cause = cause;
+ }
+
+ @Override
+ public JobStatus oldStatus() {
+ return oldStatus;
+ }
+
+ @Override
+ public JobStatus newStatus() {
+ return newStatus;
+ }
+
+ @Nullable
+ @Override
+ public Throwable exception() {
+ return cause;
+ }
+
+ @Override
+ public JobID jobId() {
+ return jobId;
+ }
+
+ @Override
+ public String jobName() {
+ return jobName;
+ }
+}
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
b/flink-core/src/main/java/org/apache/flink/core/execution/JobExecutionStatusEvent.java
similarity index 55%
copy from
flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
copy to
flink-core/src/main/java/org/apache/flink/core/execution/JobExecutionStatusEvent.java
index 761d3cc9141..82aeda5c6df 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
+++
b/flink-core/src/main/java/org/apache/flink/core/execution/JobExecutionStatusEvent.java
@@ -16,21 +16,23 @@
* limitations under the License.
*/
-package org.apache.flink.client.deployment.executors;
+package org.apache.flink.core.execution;
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.client.deployment.StandaloneClientFactory;
-import org.apache.flink.client.deployment.StandaloneClusterId;
-import org.apache.flink.core.execution.PipelineExecutor;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobStatus;
-/** The {@link PipelineExecutor} to be used when executing a job on an already
running cluster. */
-@Internal
-public class RemoteExecutor
- extends AbstractSessionClusterExecutor<StandaloneClusterId,
StandaloneClientFactory> {
+import javax.annotation.Nullable;
- public static final String NAME = "remote";
+/** Job execution status event. */
+@PublicEvolving
+public interface JobExecutionStatusEvent extends JobStatusChangedEvent {
+ /** Old status for job. */
+ JobStatus oldStatus();
- public RemoteExecutor() {
- super(new StandaloneClientFactory());
- }
+ /** New status for job. */
+ JobStatus newStatus();
+
+ /** Exception for job. */
+ @Nullable
+ Throwable exception();
}
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
b/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedEvent.java
similarity index 55%
copy from
flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
copy to
flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedEvent.java
index 761d3cc9141..21d0717ed28 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
+++
b/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedEvent.java
@@ -16,21 +16,16 @@
* limitations under the License.
*/
-package org.apache.flink.client.deployment.executors;
+package org.apache.flink.core.execution;
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.client.deployment.StandaloneClientFactory;
-import org.apache.flink.client.deployment.StandaloneClusterId;
-import org.apache.flink.core.execution.PipelineExecutor;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobID;
-/** The {@link PipelineExecutor} to be used when executing a job on an already
running cluster. */
-@Internal
-public class RemoteExecutor
- extends AbstractSessionClusterExecutor<StandaloneClusterId,
StandaloneClientFactory> {
+/** Basic job status event. */
+@PublicEvolving
+public interface JobStatusChangedEvent {
- public static final String NAME = "remote";
+ JobID jobId();
- public RemoteExecutor() {
- super(new StandaloneClientFactory());
- }
+ String jobName();
}
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
b/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListener.java
similarity index 55%
copy from
flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
copy to
flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListener.java
index 761d3cc9141..3610cd36055 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
+++
b/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListener.java
@@ -16,21 +16,17 @@
* limitations under the License.
*/
-package org.apache.flink.client.deployment.executors;
+package org.apache.flink.core.execution;
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.client.deployment.StandaloneClientFactory;
-import org.apache.flink.client.deployment.StandaloneClusterId;
-import org.apache.flink.core.execution.PipelineExecutor;
+import org.apache.flink.annotation.PublicEvolving;
-/** The {@link PipelineExecutor} to be used when executing a job on an already
running cluster. */
-@Internal
-public class RemoteExecutor
- extends AbstractSessionClusterExecutor<StandaloneClusterId,
StandaloneClientFactory> {
-
- public static final String NAME = "remote";
+/**
+ * When job is created or its status is changed, Flink will generate job event
and notify job status
+ * changed listener.
+ */
+@PublicEvolving
+public interface JobStatusChangedListener {
- public RemoteExecutor() {
- super(new StandaloneClientFactory());
- }
+ /* Event will be fired when job status is changed. */
+ void onEvent(JobStatusChangedEvent event);
}
diff --git
a/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListenerFactory.java
b/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListenerFactory.java
new file mode 100644
index 00000000000..03c1d49a0d6
--- /dev/null
+++
b/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListenerFactory.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.execution;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.Configuration;
+
+import java.util.concurrent.Executor;
+
+/** Factory for job status changed listener. */
+@PublicEvolving
+public interface JobStatusChangedListenerFactory {
+
+ JobStatusChangedListener createListener(Context context);
+
+ @PublicEvolving
+ interface Context {
+ /*
+ * Configuration for the factory to create listener, users can add
customized options to flink and get them here to create the listener. For
+ * example, users can add rest address for datahub to the
configuration, and get it when they need to create http client for the listener.
+ */
+ Configuration getConfiguration();
+
+ /**
+ * User classloader for the flink application.
+ *
+ * @return
+ */
+ ClassLoader getUserClassLoader();
+
+ /*
+ * Get an Executor pool for the listener to run async operations that
can potentially be IO-heavy. `JobMaster` will provide an independent executor
+ * for io operations and it won't block the main-thread. All tasks
submitted to the executor will be executed in parallel, and when the job ends,
+ * previously submitted tasks will be executed, but no new tasks will
be accepted.
+ */
+ Executor getIOExecutor();
+ }
+}
diff --git
a/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListenerUtils.java
b/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListenerUtils.java
new file mode 100644
index 00000000000..a73641d7ffd
--- /dev/null
+++
b/flink-core/src/main/java/org/apache/flink/core/execution/JobStatusChangedListenerUtils.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
+
+import static
org.apache.flink.configuration.DeploymentOptions.JOB_STATUS_CHANGED_LISTENERS;
+
+/** Util class for {@link JobStatusChangedListener}. */
+@Internal
+public final class JobStatusChangedListenerUtils {
+ /**
+ * Create job status changed listeners from configuration for job.
+ *
+ * @param configuration The job configuration.
+ * @return the job status changed listeners.
+ */
+ public static List<JobStatusChangedListener>
createJobStatusChangedListeners(
+ ClassLoader userClassLoader, Configuration configuration, Executor
ioExecutor) {
+ List<String> jobStatusChangedListeners =
configuration.get(JOB_STATUS_CHANGED_LISTENERS);
+ if (jobStatusChangedListeners == null ||
jobStatusChangedListeners.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return jobStatusChangedListeners.stream()
+ .map(
+ fac -> {
+ try {
+ return InstantiationUtil.instantiate(
+ fac,
+
JobStatusChangedListenerFactory.class,
+ userClassLoader)
+ .createListener(
+ new
JobStatusChangedListenerFactory.Context() {
+ @Override
+ public Configuration
getConfiguration() {
+ return configuration;
+ }
+
+ @Override
+ public ClassLoader
getUserClassLoader() {
+ return userClassLoader;
+ }
+
+ @Override
+ public Executor
getIOExecutor() {
+ return ioExecutor;
+ }
+ });
+ } catch (FlinkException e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .collect(Collectors.toList());
+ }
+}
diff --git
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/executors/KubernetesSessionClusterExecutor.java
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/executors/KubernetesSessionClusterExecutor.java
index c9edef90777..84eba0e1565 100644
---
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/executors/KubernetesSessionClusterExecutor.java
+++
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/executors/KubernetesSessionClusterExecutor.java
@@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.executors;
import org.apache.flink.annotation.Internal;
import
org.apache.flink.client.deployment.executors.AbstractSessionClusterExecutor;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.PipelineExecutor;
import org.apache.flink.kubernetes.KubernetesClusterClientFactory;
import org.apache.flink.kubernetes.configuration.KubernetesDeploymentTarget;
@@ -31,7 +32,7 @@ public class KubernetesSessionClusterExecutor
public static final String NAME =
KubernetesDeploymentTarget.SESSION.getName();
- public KubernetesSessionClusterExecutor() {
- super(new KubernetesClusterClientFactory());
+ public KubernetesSessionClusterExecutor(Configuration configuration) {
+ super(new KubernetesClusterClientFactory(), configuration);
}
}
diff --git
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/executors/KubernetesSessionClusterExecutorFactory.java
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/executors/KubernetesSessionClusterExecutorFactory.java
index 90c6a62adf5..196bc201039 100644
---
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/executors/KubernetesSessionClusterExecutorFactory.java
+++
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/executors/KubernetesSessionClusterExecutorFactory.java
@@ -44,6 +44,6 @@ public class KubernetesSessionClusterExecutorFactory
implements PipelineExecutor
@Override
public PipelineExecutor getExecutor(@Nonnull final Configuration
configuration) {
- return new KubernetesSessionClusterExecutor();
+ return new KubernetesSessionClusterExecutor(configuration);
}
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
index c53e227ec1e..afd36a2951f 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
@@ -27,6 +27,8 @@ import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.accumulators.AccumulatorHelper;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.DefaultJobExecutionStatusEvent;
+import org.apache.flink.core.execution.JobStatusChangedListener;
import org.apache.flink.core.execution.JobStatusHook;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.SimpleCounter;
@@ -302,6 +304,8 @@ public class DefaultExecutionGraph implements
ExecutionGraph, InternalExecutionG
private final TaskDeploymentDescriptorFactory
taskDeploymentDescriptorFactory;
+ private final List<JobStatusChangedListener> jobStatusChangedListeners;
+
//
--------------------------------------------------------------------------------------------
// Constructors
//
--------------------------------------------------------------------------------------------
@@ -327,7 +331,8 @@ public class DefaultExecutionGraph implements
ExecutionGraph, InternalExecutionG
ExecutionJobVertex.Factory executionJobVertexFactory,
List<JobStatusHook> jobStatusHooks,
MarkPartitionFinishedStrategy markPartitionFinishedStrategy,
- TaskDeploymentDescriptorFactory taskDeploymentDescriptorFactory) {
+ TaskDeploymentDescriptorFactory taskDeploymentDescriptorFactory,
+ List<JobStatusChangedListener> jobStatusChangedListeners) {
this.jobType = jobType;
this.executionGraphId = new ExecutionGraphID();
@@ -398,6 +403,8 @@ public class DefaultExecutionGraph implements
ExecutionGraph, InternalExecutionG
this.taskDeploymentDescriptorFactory =
checkNotNull(taskDeploymentDescriptorFactory);
+ this.jobStatusChangedListeners =
checkNotNull(jobStatusChangedListeners);
+
LOG.info(
"Created execution graph {} for job {}.",
executionGraphId,
@@ -1164,7 +1171,7 @@ public class DefaultExecutionGraph implements
ExecutionGraph, InternalExecutionG
error);
stateTimestamps[newState.ordinal()] = System.currentTimeMillis();
- notifyJobStatusChange(newState);
+ notifyJobStatusChange(current, newState, error);
notifyJobStatusHooks(newState, error);
return true;
} else {
@@ -1600,7 +1607,8 @@ public class DefaultExecutionGraph implements
ExecutionGraph, InternalExecutionG
}
}
- private void notifyJobStatusChange(JobStatus newState) {
+ private void notifyJobStatusChange(
+ JobStatus oldState, JobStatus newState, @Nullable Throwable cause)
{
if (jobStatusListeners.size() > 0) {
final long timestamp = System.currentTimeMillis();
@@ -1612,6 +1620,14 @@ public class DefaultExecutionGraph implements
ExecutionGraph, InternalExecutionG
}
}
}
+
+ if (jobStatusChangedListeners.size() > 0) {
+ jobStatusChangedListeners.forEach(
+ listener ->
+ listener.onEvent(
+ new DefaultJobExecutionStatusEvent(
+ getJobID(), getJobName(),
oldState, newState, cause)));
+ }
}
private void notifyJobStatusHooks(JobStatus newState, @Nullable Throwable
cause) {
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
index 57be7f49abf..30b5af6e3f7 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
@@ -23,6 +23,8 @@ import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.core.execution.JobStatusChangedListener;
+import org.apache.flink.core.execution.JobStatusChangedListenerUtils;
import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.blob.BlobWriter;
import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
@@ -141,6 +143,10 @@ public class DefaultExecutionGraphBuilder {
throw new JobException("Could not create the
TaskDeploymentDescriptorFactory.", e);
}
+ final List<JobStatusChangedListener> jobStatusChangedListeners =
+ JobStatusChangedListenerUtils.createJobStatusChangedListeners(
+ classLoader, jobManagerConfig, ioExecutor);
+
// create a new execution graph, if none exists so far
final DefaultExecutionGraph executionGraph =
new DefaultExecutionGraph(
@@ -164,7 +170,8 @@ public class DefaultExecutionGraphBuilder {
executionJobVertexFactory,
jobGraph.getJobStatusHooks(),
markPartitionFinishedStrategy,
- taskDeploymentDescriptorFactory);
+ taskDeploymentDescriptorFactory,
+ jobStatusChangedListeners);
// set the basic properties
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
index fb1deadff6e..408bf00fd11 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
@@ -46,6 +46,7 @@ import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.lineage.LineageGraph;
import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
import org.apache.flink.streaming.api.operators.OutputFormatOperatorFactory;
import org.apache.flink.streaming.api.operators.SourceOperatorFactory;
@@ -134,6 +135,7 @@ public class StreamGraph implements Pipeline {
private boolean dynamic;
private boolean autoParallelismEnabled;
+ private LineageGraph lineageGraph;
public StreamGraph(
Configuration jobConfiguration,
@@ -228,6 +230,10 @@ public class StreamGraph implements Pipeline {
this.timeCharacteristic = timeCharacteristic;
}
+ public LineageGraph getLineageGraph() {
+ return lineageGraph;
+ }
+
public GlobalStreamExchangeMode getGlobalStreamExchangeMode() {
return globalExchangeMode;
}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/execution/DefaultJobCreatedEvent.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/execution/DefaultJobCreatedEvent.java
new file mode 100644
index 00000000000..34ff064137c
--- /dev/null
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/execution/DefaultJobCreatedEvent.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.streaming.api.lineage.LineageGraph;
+
+/** Default implementation for {@link JobCreatedEvent}. */
+@Internal
+public class DefaultJobCreatedEvent implements JobCreatedEvent {
+ private final JobID jobId;
+ private final String jobName;
+ private final LineageGraph lineageGraph;
+ private final RuntimeExecutionMode executionMode;
+
+ public DefaultJobCreatedEvent(
+ JobID jobId,
+ String jobName,
+ LineageGraph lineageGraph,
+ RuntimeExecutionMode executionMode) {
+ this.jobId = jobId;
+ this.jobName = jobName;
+ this.lineageGraph = lineageGraph;
+ this.executionMode = executionMode;
+ }
+
+ @Override
+ public JobID jobId() {
+ return jobId;
+ }
+
+ @Override
+ public String jobName() {
+ return jobName;
+ }
+
+ @Override
+ public LineageGraph lineageGraph() {
+ return lineageGraph;
+ }
+
+ @Override
+ public RuntimeExecutionMode executionMode() {
+ return executionMode;
+ }
+}
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/execution/JobCreatedEvent.java
similarity index 55%
copy from
flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
copy to
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/execution/JobCreatedEvent.java
index 761d3cc9141..1d20dcd2108 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/RemoteExecutor.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/execution/JobCreatedEvent.java
@@ -16,21 +16,20 @@
* limitations under the License.
*/
-package org.apache.flink.client.deployment.executors;
+package org.apache.flink.streaming.runtime.execution;
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.client.deployment.StandaloneClientFactory;
-import org.apache.flink.client.deployment.StandaloneClusterId;
-import org.apache.flink.core.execution.PipelineExecutor;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.core.execution.JobStatusChangedEvent;
+import org.apache.flink.streaming.api.lineage.LineageGraph;
-/** The {@link PipelineExecutor} to be used when executing a job on an already
running cluster. */
-@Internal
-public class RemoteExecutor
- extends AbstractSessionClusterExecutor<StandaloneClusterId,
StandaloneClientFactory> {
+/** Basic job created event. */
+@PublicEvolving
+public interface JobCreatedEvent extends JobStatusChangedEvent {
- public static final String NAME = "remote";
+ /* Lineage for the current job. */
+ LineageGraph lineageGraph();
- public RemoteExecutor() {
- super(new StandaloneClientFactory());
- }
+ /* Runtime execution mode for the job, STREAMING/BATCH/AUTOMATIC. */
+ RuntimeExecutionMode executionMode();
}
diff --git
a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/MiniClusterPipelineExecutorServiceLoader.java
b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/MiniClusterPipelineExecutorServiceLoader.java
index c86639f76b3..ddfe8a1a8f4 100644
---
a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/MiniClusterPipelineExecutorServiceLoader.java
+++
b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/MiniClusterPipelineExecutorServiceLoader.java
@@ -27,6 +27,8 @@ import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.core.execution.CacheSupportedPipelineExecutor;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobStatusChangedListener;
+import org.apache.flink.core.execution.JobStatusChangedListenerUtils;
import org.apache.flink.core.execution.PipelineExecutor;
import org.apache.flink.core.execution.PipelineExecutorFactory;
import org.apache.flink.core.execution.PipelineExecutorServiceLoader;
@@ -38,6 +40,7 @@ import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.minicluster.MiniClusterJobClient;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.util.AbstractID;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,8 +49,11 @@ import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Collection;
+import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.stream.Stream;
/**
@@ -145,11 +151,19 @@ public class MiniClusterPipelineExecutorServiceLoader
implements PipelineExecuto
}
private static class MiniClusterExecutor implements
CacheSupportedPipelineExecutor {
-
+ private final ExecutorService executorService =
+ Executors.newFixedThreadPool(
+ 1, new
ExecutorThreadFactory("Flink-MiniClusterExecutor-IO"));
private final MiniCluster miniCluster;
+ private final List<JobStatusChangedListener> jobStatusChangedListeners;
public MiniClusterExecutor(MiniCluster miniCluster) {
this.miniCluster = miniCluster;
+ this.jobStatusChangedListeners =
+
JobStatusChangedListenerUtils.createJobStatusChangedListeners(
+ Thread.currentThread().getContextClassLoader(),
+ miniCluster.getConfiguration(),
+ executorService);
}
@Override
@@ -165,6 +179,17 @@ public class MiniClusterPipelineExecutorServiceLoader
implements PipelineExecuto
}
return miniCluster
.submitJob(jobGraph)
+ .whenComplete(
+ (ignored, throwable) -> {
+ if (throwable == null) {
+
PipelineExecutorUtils.notifyJobStatusListeners(
+ pipeline, jobGraph,
jobStatusChangedListeners);
+ } else {
+ LOG.error(
+ "Failed to submit job graph to
mini cluster.",
+ throwable);
+ }
+ })
.thenApply(
result ->
new MiniClusterJobClient(
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/execution/JobStatusChangedListenerITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/execution/JobStatusChangedListenerITCase.java
new file mode 100644
index 00000000000..dd58ef184dc
--- /dev/null
+++
b/flink-tests/src/test/java/org/apache/flink/test/execution/JobStatusChangedListenerITCase.java
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.execution;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.DefaultJobExecutionStatusEvent;
+import org.apache.flink.core.execution.JobExecutionStatusEvent;
+import org.apache.flink.core.execution.JobStatusChangedEvent;
+import org.apache.flink.core.execution.JobStatusChangedListener;
+import org.apache.flink.core.execution.JobStatusChangedListenerFactory;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.streaming.runtime.execution.DefaultJobCreatedEvent;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runners.MethodSorters;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static
org.apache.flink.configuration.DeploymentOptions.JOB_STATUS_CHANGED_LISTENERS;
+import static
org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for job status changed listener. */
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class JobStatusChangedListenerITCase extends TestLogger {
+ private static final int PARALLELISM = 4;
+ @ClassRule public static final TemporaryFolder TMP_FOLDER = new
TemporaryFolder();
+
+ @ClassRule
+ public static final MiniClusterWithClientResource MINI_CLUSTER =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(createConfiguration())
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(PARALLELISM)
+ .build());
+
+ private static List<JobStatusChangedEvent> statusChangedEvents = new
ArrayList<>();
+
+ @Before
+ public void setup() {
+ statusChangedEvents.clear();
+ }
+
+ @Test
+ public void testJobStatusChangedForSucceededApplication() throws Exception
{
+ Configuration configuration = createConfiguration();
+ try (StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.getExecutionEnvironment(configuration)) {
+ List<String> sourceValues = Arrays.asList("a", "b", "c");
+ List<String> resultValues = new ArrayList<>();
+ try (CloseableIterator<String> iterator =
+ env.fromCollection(sourceValues).executeAndCollect()) {
+ while (iterator.hasNext()) {
+ resultValues.add(iterator.next());
+ }
+ }
+
assertThat(resultValues).containsExactlyInAnyOrder(sourceValues.toArray(new
String[0]));
+ }
+
+ verifyEventMetaData();
+ statusChangedEvents.forEach(
+ event -> {
+ if (event instanceof DefaultJobExecutionStatusEvent) {
+ JobExecutionStatusEvent status =
(JobExecutionStatusEvent) event;
+ assertThat(
+ (status.oldStatus() ==
JobStatus.CREATED
+ && status.newStatus()
== JobStatus.RUNNING)
+ || (status.oldStatus() ==
JobStatus.RUNNING
+ && status.newStatus()
+ ==
JobStatus.FINISHED))
+ .isTrue();
+ } else {
+ DefaultJobCreatedEvent createdEvent =
(DefaultJobCreatedEvent) event;
+ assertThat(createdEvent.executionMode())
+ .isEqualTo(RuntimeExecutionMode.STREAMING);
+ }
+ });
+ }
+
+ @Test
+ public void testJobStatusChangedForFailedApplication() throws Exception {
+ Configuration configuration = createConfiguration();
+
+ try (StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.getExecutionEnvironment(configuration)) {
+ env.setParallelism(PARALLELISM);
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+ env.addSource(new FastFailureSourceFunction()).addSink(new
SleepingSink());
+
+ StreamGraph streamGraph = env.getStreamGraph();
+ JobGraph jobGraph = streamGraph.getJobGraph();
+
+ ClusterClient<?> client = MINI_CLUSTER.getClusterClient();
+ JobID jobID = client.submitJob(jobGraph).get();
+ while (!client.getJobStatus(jobID).get().equals(JobStatus.FAILED))
{}
+ } catch (Exception e) {
+ // Expected failure due to exception.
+ }
+
+ verifyEventMetaData();
+ statusChangedEvents.forEach(
+ event -> {
+ JobExecutionStatusEvent status = (JobExecutionStatusEvent)
event;
+ assertThat(
+ (status.oldStatus() == JobStatus.CREATED
+ && status.newStatus() ==
JobStatus.RUNNING)
+ || (status.oldStatus() ==
JobStatus.RUNNING
+ && status.newStatus() ==
JobStatus.FAILING)
+ || (status.oldStatus() ==
JobStatus.FAILING
+ && status.newStatus() ==
JobStatus.FAILED))
+ .isTrue();
+ });
+ }
+
+ @Test
+ public void testJobStatusChangedForCancelledApplication() throws Exception
{
+ Configuration configuration = createConfiguration();
+
+ try (StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.getExecutionEnvironment(configuration)) {
+ final DataStream<Long> source = env.addSource(new
InfiniteLongSourceFunction());
+ source.addSink(new SleepingSink());
+
+ StreamGraph streamGraph = env.getStreamGraph();
+ JobGraph jobGraph = streamGraph.getJobGraph();
+
+ ClusterClient<?> client = MINI_CLUSTER.getClusterClient();
+ JobID jobID = client.submitJob(jobGraph).get();
+ waitForAllTaskRunning(MINI_CLUSTER.getMiniCluster(), jobID, false);
+
+ Thread.sleep(100);
+ client.cancel(jobID).get();
+ while
(!client.getJobStatus(jobID).get().equals(JobStatus.CANCELED)) {}
+ }
+
+ verifyEventMetaData();
+ statusChangedEvents.forEach(
+ event -> {
+ JobExecutionStatusEvent status = (JobExecutionStatusEvent)
event;
+ assertThat(
+ (status.oldStatus() == JobStatus.CREATED
+ && status.newStatus() ==
JobStatus.RUNNING)
+ || (status.oldStatus() ==
JobStatus.RUNNING
+ && status.newStatus() ==
JobStatus.CANCELLING)
+ || (status.oldStatus() ==
JobStatus.CANCELLING
+ && status.newStatus() ==
JobStatus.CANCELED))
+ .isTrue();
+ });
+ }
+
+ void verifyEventMetaData() {
+ assertThat(statusChangedEvents.size()).isEqualTo(3);
+ assertThat(statusChangedEvents.get(0).jobId())
+ .isEqualTo(statusChangedEvents.get(1).jobId());
+ assertThat(statusChangedEvents.get(0).jobName())
+ .isEqualTo(statusChangedEvents.get(1).jobName());
+
+ assertThat(statusChangedEvents.get(1).jobId())
+ .isEqualTo(statusChangedEvents.get(2).jobId());
+ assertThat(statusChangedEvents.get(1).jobName())
+ .isEqualTo(statusChangedEvents.get(2).jobName());
+ }
+
+ /** Testing job status changed listener factory. */
+ public static class TestingJobStatusChangedListenerFactory
+ implements JobStatusChangedListenerFactory {
+
+ @Override
+ public JobStatusChangedListener createListener(Context context) {
+ return new TestingJobStatusChangedListener();
+ }
+ }
+
+ /** Testing job status changed listener. */
+ private static class TestingJobStatusChangedListener implements
JobStatusChangedListener {
+
+ @Override
+ public void onEvent(JobStatusChangedEvent event) {
+ statusChangedEvents.add(event);
+ }
+ }
+
+ private static Configuration createConfiguration() {
+ Configuration configuration = new Configuration();
+ configuration.set(
+ JOB_STATUS_CHANGED_LISTENERS,
+
Collections.singletonList(TestingJobStatusChangedListenerFactory.class.getName()));
+
+ return configuration;
+ }
+
+ private static class FastFailureSourceFunction implements
SourceFunction<Long> {
+
+ @Override
+ public void run(SourceContext<Long> ctx) throws Exception {
+ throw new RuntimeException("Failed to execute.");
+ }
+
+ @Override
+ public void cancel() {}
+ }
+
+ private static class InfiniteLongSourceFunction implements
SourceFunction<Long> {
+ private volatile boolean running = true;
+
+ @Override
+ public void run(SourceContext<Long> ctx) throws Exception {
+ long next = 0;
+ while (running) {
+ synchronized (ctx.getCheckpointLock()) {
+ ctx.collect(next++);
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ running = false;
+ }
+ }
+
+ private static class SleepingSink implements SinkFunction<Long> {
+ @Override
+ public void invoke(Long value, Context context) throws Exception {
+ Thread.sleep(1_000);
+ }
+ }
+}
diff --git
a/flink-yarn/src/main/java/org/apache/flink/yarn/executors/YarnSessionClusterExecutor.java
b/flink-yarn/src/main/java/org/apache/flink/yarn/executors/YarnSessionClusterExecutor.java
index b14618e0515..065772daee8 100644
---
a/flink-yarn/src/main/java/org/apache/flink/yarn/executors/YarnSessionClusterExecutor.java
+++
b/flink-yarn/src/main/java/org/apache/flink/yarn/executors/YarnSessionClusterExecutor.java
@@ -20,6 +20,7 @@ package org.apache.flink.yarn.executors;
import org.apache.flink.annotation.Internal;
import
org.apache.flink.client.deployment.executors.AbstractSessionClusterExecutor;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.PipelineExecutor;
import org.apache.flink.yarn.YarnClusterClientFactory;
import org.apache.flink.yarn.configuration.YarnDeploymentTarget;
@@ -33,7 +34,7 @@ public class YarnSessionClusterExecutor
public static final String NAME = YarnDeploymentTarget.SESSION.getName();
- public YarnSessionClusterExecutor() {
- super(new YarnClusterClientFactory());
+ public YarnSessionClusterExecutor(Configuration configuration) {
+ super(new YarnClusterClientFactory(), configuration);
}
}
diff --git
a/flink-yarn/src/main/java/org/apache/flink/yarn/executors/YarnSessionClusterExecutorFactory.java
b/flink-yarn/src/main/java/org/apache/flink/yarn/executors/YarnSessionClusterExecutorFactory.java
index e94a230b4cd..3c7cb0f9044 100644
---
a/flink-yarn/src/main/java/org/apache/flink/yarn/executors/YarnSessionClusterExecutorFactory.java
+++
b/flink-yarn/src/main/java/org/apache/flink/yarn/executors/YarnSessionClusterExecutorFactory.java
@@ -45,7 +45,7 @@ public class YarnSessionClusterExecutorFactory implements
PipelineExecutorFactor
@Override
public PipelineExecutor getExecutor(@Nonnull final Configuration
configuration) {
try {
- return new YarnSessionClusterExecutor();
+ return new YarnSessionClusterExecutor(configuration);
} catch (NoClassDefFoundError e) {
throw new
IllegalStateException(YarnDeploymentTarget.ERROR_MESSAGE);
}
