You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/12/10 08:05:26 UTC
[flink] 03/04: [FLINK-14992][client] Add job listener to execution
environments
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5afc3084aac7ce89f1648926898fea2c73d439cb
Author: tison <wa...@gmail.com>
AuthorDate: Wed Dec 4 09:28:43 2019 +0800
[FLINK-14992][client] Add job listener to execution environments
---
.../apache/flink/core/execution/JobListener.java | 57 ++++++++++++++
.../flink/api/java/ExecutionEnvironment.java | 53 +++++++++++--
.../test_execution_environment_completeness.py | 2 +-
...st_stream_execution_environment_completeness.py | 3 +-
.../flink/api/scala/ExecutionEnvironment.scala | 18 ++++-
.../environment/StreamExecutionEnvironment.java | 56 +++++++++++--
.../api/scala/StreamExecutionEnvironment.scala | 19 ++++-
.../flink/test/execution/JobListenerITCase.java | 91 ++++++++++++++++++++++
8 files changed, 285 insertions(+), 14 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/core/execution/JobListener.java b/flink-core/src/main/java/org/apache/flink/core/execution/JobListener.java
new file mode 100644
index 0000000..e67db44
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/execution/JobListener.java
@@ -0,0 +1,57 @@
+/*
+ * 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.api.common.JobExecutionResult;
+
+import javax.annotation.Nullable;
+
+/**
+ * A listener that is notified on specific job status changed, which should be firstly
+ * registered by {@code #registerJobListener} of execution environments.
+ *
+ * <p>It is highly recommended NOT to perform any blocking operation inside
+ * the callbacks. If you block the thread the invoker of environment execute methods
+ * is possibly blocked.
+ */
+@PublicEvolving
+public interface JobListener {
+
+ /**
+ * Callback on job submission. This is called when {@code execute()} or {@code executeAsync()}
+ * is called.
+ *
+ * <p>Exactly one of the passed parameters is null, respectively for failure or success.
+ *
+ * @param jobClient a {@link JobClient} for the submitted Flink job
+ * @param throwable the cause if submission failed
+ */
+ void onJobSubmitted(@Nullable JobClient jobClient, @Nullable Throwable throwable);
+
+ /**
+ * Callback on job execution finished, successfully or unsuccessfully. It is only called
+ * back when you call {@code execute()} instead of {@code executeAsync()} methods of execution
+ * environments.
+ *
+ * <p>Exactly one of the passed parameters is null, respectively for failure or success.
+ */
+ void onJobExecuted(@Nullable JobExecutionResult jobExecutionResult, @Nullable Throwable throwable);
+
+}
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
index cf82c22..6078470 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
@@ -58,8 +58,10 @@ import org.apache.flink.core.execution.DetachedJobExecutionResult;
import org.apache.flink.core.execution.ExecutorFactory;
import org.apache.flink.core.execution.ExecutorServiceLoader;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobListener;
import org.apache.flink.core.fs.Path;
import org.apache.flink.types.StringValue;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.NumberSequenceIterator;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SplittableIterator;
@@ -136,6 +138,8 @@ public class ExecutionEnvironment {
private final ClassLoader userClassloader;
+ private final List<JobListener> jobListeners = new ArrayList<>();
+
/**
* Creates a new {@link ExecutionEnvironment} that will use the given {@link Configuration} to
* configure the {@link org.apache.flink.core.execution.Executor}.
@@ -819,16 +823,45 @@ public class ExecutionEnvironment {
public JobExecutionResult execute(String jobName) throws Exception {
final JobClient jobClient = executeAsync(jobName);
- if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
- lastJobExecutionResult = jobClient.getJobExecutionResult(userClassloader).get();
- } else {
- lastJobExecutionResult = new DetachedJobExecutionResult(jobClient.getJobID());
+ try {
+ if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
+ lastJobExecutionResult = jobClient.getJobExecutionResult(userClassloader).get();
+ } else {
+ lastJobExecutionResult = new DetachedJobExecutionResult(jobClient.getJobID());
+ }
+
+ jobListeners.forEach(
+ jobListener -> jobListener.onJobExecuted(lastJobExecutionResult, null));
+
+ } catch (Throwable t) {
+ jobListeners.forEach(jobListener -> {
+ jobListener.onJobExecuted(null, ExceptionUtils.stripExecutionException(t));
+ });
+ ExceptionUtils.rethrowException(t);
}
return lastJobExecutionResult;
}
/**
+ * Register a {@link JobListener} in this environment. The {@link JobListener} will be
+ * notified on specific job status changed.
+ */
+ @PublicEvolving
+ public void registerJobListener(JobListener jobListener) {
+ checkNotNull(jobListener, "JobListener cannot be null");
+ jobListeners.add(jobListener);
+ }
+
+ /**
+ * Clear all registered {@link JobListener}s.
+ */
+ @PublicEvolving
+ public void clearJobListeners() {
+ this.jobListeners.clear();
+ }
+
+ /**
* Triggers the program execution asynchronously. The environment will execute all parts of the program that have
* resulted in a "sink" operation. Sink operations are for example printing results ({@link DataSet#print()},
* writing results (e.g. {@link DataSet#writeAsText(String)},
@@ -876,7 +909,17 @@ public class ExecutionEnvironment {
.getExecutor(configuration)
.execute(plan, configuration);
- return jobClientFuture.get();
+ try {
+ JobClient jobClient = jobClientFuture.get();
+ jobListeners.forEach(jobListener -> jobListener.onJobSubmitted(jobClient, null));
+ return jobClient;
+ } catch (Throwable t) {
+ jobListeners.forEach(jobListener -> jobListener.onJobSubmitted(null, t));
+ ExceptionUtils.rethrow(t);
+
+ // make javac happy, this code path will not be reached
+ return null;
+ }
}
private void consolidateParallelismDefinitionsInConfiguration() {
diff --git a/flink-python/pyflink/dataset/tests/test_execution_environment_completeness.py b/flink-python/pyflink/dataset/tests/test_execution_environment_completeness.py
index e4dbe29..02f83fd6 100644
--- a/flink-python/pyflink/dataset/tests/test_execution_environment_completeness.py
+++ b/flink-python/pyflink/dataset/tests/test_execution_environment_completeness.py
@@ -51,7 +51,7 @@ class ExecutionEnvironmentCompletenessTests(PythonAPICompletenessTestCase,
'createCollectionsEnvironment', 'readFile', 'readFileOfPrimitives',
'generateSequence', 'areExplicitEnvironmentsAllowed', 'createInput',
'getUserCodeClassLoader', 'getExecutorServiceLoader', 'getConfiguration',
- 'executeAsync'}
+ 'executeAsync', 'registerJobListener', 'clearJobListeners'}
if __name__ == '__main__':
diff --git a/flink-python/pyflink/datastream/tests/test_stream_execution_environment_completeness.py b/flink-python/pyflink/datastream/tests/test_stream_execution_environment_completeness.py
index 7cfa605..a1a325f 100644
--- a/flink-python/pyflink/datastream/tests/test_stream_execution_environment_completeness.py
+++ b/flink-python/pyflink/datastream/tests/test_stream_execution_environment_completeness.py
@@ -48,7 +48,8 @@ class StreamExecutionEnvironmentCompletenessTests(PythonAPICompletenessTestCase,
'readFileStream', 'isForceCheckpointing', 'readFile', 'clean',
'createInput', 'createLocalEnvironmentWithWebUI', 'fromCollection',
'socketTextStream', 'initializeContextEnvironment', 'readTextFile', 'addSource',
- 'setNumberOfExecutionRetries', 'configure', 'executeAsync'}
+ 'setNumberOfExecutionRetries', 'configure', 'executeAsync', 'registerJobListener',
+ 'clearJobListeners'}
if __name__ == '__main__':
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
index 9c5598d..79d32d7 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
@@ -32,7 +32,7 @@ import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfoBase, ValueTypeInfo}
import org.apache.flink.api.java.{CollectionEnvironment, ExecutionEnvironment => JavaEnv}
import org.apache.flink.configuration.Configuration
-import org.apache.flink.core.execution.JobClient
+import org.apache.flink.core.execution.{JobClient, JobListener}
import org.apache.flink.core.fs.Path
import org.apache.flink.types.StringValue
import org.apache.flink.util.{NumberSequenceIterator, Preconditions, SplittableIterator}
@@ -495,6 +495,22 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
}
/**
+ * Register a [[JobListener]] in this environment. The [[JobListener]] will be
+ * notified on specific job status changed.
+ */
+ @PublicEvolving
+ def registerJobListener(jobListener: JobListener): Unit = {
+ javaEnv.registerJobListener(jobListener)
+ }
+
+ /**
+ * Clear all registered [[JobListener]]s.
+ */
+ @PublicEvolving def clearJobListeners(): Unit = {
+ javaEnv.clearJobListeners()
+ }
+
+ /**
* Triggers the program execution asynchronously.
* The environment will execute all parts of the program that have
* resulted in a "sink" operation. Sink operations are for example printing results
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index 34aab2e..dc1053f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -58,6 +58,7 @@ import org.apache.flink.core.execution.DetachedJobExecutionResult;
import org.apache.flink.core.execution.ExecutorFactory;
import org.apache.flink.core.execution.ExecutorServiceLoader;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobListener;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
@@ -85,6 +86,7 @@ import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
import org.apache.flink.streaming.api.operators.StreamSource;
import org.apache.flink.util.DynamicCodeLoadingException;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SplittableIterator;
import org.apache.flink.util.StringUtils;
@@ -167,6 +169,8 @@ public class StreamExecutionEnvironment {
private final ClassLoader userClassloader;
+ private final List<JobListener> jobListeners = new ArrayList<>();
+
// --------------------------------------------------------------------------------------------
// Constructor and Properties
// --------------------------------------------------------------------------------------------
@@ -1637,14 +1641,46 @@ public class StreamExecutionEnvironment {
public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
final JobClient jobClient = executeAsync(streamGraph);
- if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
- return jobClient.getJobExecutionResult(userClassloader).get();
- } else {
- return new DetachedJobExecutionResult(jobClient.getJobID());
+ try {
+ JobExecutionResult jobExecutionResult =
+ configuration.getBoolean(DeploymentOptions.ATTACHED) ?
+ jobClient.getJobExecutionResult(userClassloader).get()
+ : new DetachedJobExecutionResult(jobClient.getJobID());
+
+ jobListeners
+ .forEach(jobListener -> jobListener.onJobExecuted(jobExecutionResult, null));
+
+ return jobExecutionResult;
+ } catch (Throwable t) {
+ jobListeners.forEach(jobListener -> {
+ jobListener.onJobExecuted(null, ExceptionUtils.stripExecutionException(t));
+ });
+ ExceptionUtils.rethrowException(t);
+
+ // never reached, only make javac happy
+ return null;
}
}
/**
+ * Register a {@link JobListener} in this environment. The {@link JobListener} will be
+ * notified on specific job status changed.
+ */
+ @PublicEvolving
+ public void registerJobListener(JobListener jobListener) {
+ checkNotNull(jobListener, "JobListener cannot be null");
+ jobListeners.add(jobListener);
+ }
+
+ /**
+ * Clear all registered {@link JobListener}s.
+ */
+ @PublicEvolving
+ public void clearJobListeners() {
+ this.jobListeners.clear();
+ }
+
+ /**
* Triggers the program asynchronously. The environment will execute all parts of
* the program that have resulted in a "sink" operation. Sink operations are
* for example printing results or forwarding them to a message queue.
@@ -1704,7 +1740,17 @@ public class StreamExecutionEnvironment {
.getExecutor(configuration)
.execute(streamGraph, configuration);
- return jobClientFuture.get();
+ try {
+ JobClient jobClient = jobClientFuture.get();
+ jobListeners.forEach(jobListener -> jobListener.onJobSubmitted(jobClient, null));
+ return jobClient;
+ } catch (Throwable t) {
+ jobListeners.forEach(jobListener -> jobListener.onJobSubmitted(null, t));
+ ExceptionUtils.rethrow(t);
+
+ // make javac happy, this code path will not be reached
+ return null;
+ }
}
private void consolidateParallelismDefinitionsInConfiguration() {
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
index f911f3e..9eb0e33 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
@@ -29,13 +29,14 @@ import org.apache.flink.api.java.typeutils.ResultTypeQueryable
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
import org.apache.flink.api.scala.ClosureCleaner
import org.apache.flink.configuration.{Configuration, ReadableConfig}
-import org.apache.flink.core.execution.JobClient
+import org.apache.flink.core.execution.{JobClient, JobListener}
import org.apache.flink.runtime.state.AbstractStateBackend
import org.apache.flink.runtime.state.StateBackend
import org.apache.flink.streaming.api.environment.{StreamExecutionEnvironment => JavaEnv}
import org.apache.flink.streaming.api.functions.source._
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
+import org.apache.flink.util.Preconditions.checkNotNull
import org.apache.flink.util.SplittableIterator
import scala.collection.JavaConverters._
@@ -680,6 +681,22 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
def execute(jobName: String) = javaEnv.execute(jobName)
/**
+ * Register a [[JobListener]] in this environment. The [[JobListener]] will be
+ * notified on specific job status changed.
+ */
+ @PublicEvolving
+ def registerJobListener(jobListener: JobListener): Unit = {
+ javaEnv.registerJobListener(jobListener)
+ }
+
+ /**
+ * Clear all registered [[JobListener]]s.
+ */
+ @PublicEvolving def clearJobListeners(): Unit = {
+ javaEnv.clearJobListeners()
+ }
+
+ /**
* Triggers the program execution asynchronously. The environment will execute all parts of
* the program that have resulted in a "sink" operation. Sink operations are
* for example printing results or forwarding them to a message queue.
diff --git a/flink-tests/src/test/java/org/apache/flink/test/execution/JobListenerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/execution/JobListenerITCase.java
new file mode 100644
index 0000000..ad0173f
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/execution/JobListenerITCase.java
@@ -0,0 +1,91 @@
+/*
+ * 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.JobExecutionResult;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobListener;
+import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Unit tests for {@link JobListener}.
+ */
+public class JobListenerITCase extends TestLogger {
+
+ @Test
+ public void testJobListenerOnBatchEnvironment() throws Exception {
+ OneShotLatch submissionLatch = new OneShotLatch();
+ OneShotLatch executionLatch = new OneShotLatch();
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ env.registerJobListener(new JobListener() {
+ @Override
+ public void onJobSubmitted(JobClient jobClient, Throwable throwable) {
+ submissionLatch.trigger();
+ }
+
+ @Override
+ public void onJobExecuted(JobExecutionResult jobExecutionResult, Throwable throwable) {
+ executionLatch.trigger();
+ }
+ });
+
+ env.fromElements(1, 2, 3, 4, 5).output(new DiscardingOutputFormat<>());
+ env.execute();
+
+ submissionLatch.await(2000L, TimeUnit.MILLISECONDS);
+ executionLatch.await(2000L, TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ public void testJobListenerOnStreamingEnvironment() throws Exception {
+ OneShotLatch submissionLatch = new OneShotLatch();
+ OneShotLatch executionLatch = new OneShotLatch();
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ env.registerJobListener(new JobListener() {
+ @Override
+ public void onJobSubmitted(JobClient jobClient, Throwable throwable) {
+ submissionLatch.trigger();
+ }
+
+ @Override
+ public void onJobExecuted(JobExecutionResult jobExecutionResult, Throwable throwable) {
+ executionLatch.trigger();
+ }
+ });
+
+ env.fromElements(1, 2, 3, 4, 5).addSink(new DiscardingSink<>());
+ env.execute();
+
+ submissionLatch.await(2000L, TimeUnit.MILLISECONDS);
+ executionLatch.await(2000L, TimeUnit.MILLISECONDS);
+ }
+}