This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new ed1018e [STREAMPIPES-488] Properly shutdown Flink job in debug mode
ed1018e is described below
commit ed1018e13834deb51196e1899cc16324b1ceed02
Author: Dominik Riemer <[email protected]>
AuthorDate: Tue Dec 28 13:26:07 2021 +0100
[STREAMPIPES-488] Properly shutdown Flink job in debug mode
---
.../streampipes/wrapper/flink/FlinkRuntime.java | 35 ++++++++++++++++----
.../wrapper/flink/FlinkSpMiniCluster.java | 38 ++++++++++++++++++++++
2 files changed, 66 insertions(+), 7 deletions(-)
diff --git
a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkRuntime.java
b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkRuntime.java
index 1c652e3..b1f6883 100644
---
a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkRuntime.java
+++
b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkRuntime.java
@@ -18,6 +18,8 @@
package org.apache.streampipes.wrapper.flink;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.MiniClusterClient;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
@@ -73,8 +75,15 @@ public abstract class FlinkRuntime<RP extends
RuntimeParams<B, I, RC>, B extends
public void run() {
try {
- env.execute(bindingParams.getGraph().getElementId());
-
+ if (!this.config.isMiniClusterMode()) {
+ env.execute(bindingParams.getGraph().getElementId());
+ } else {
+ FlinkSpMiniCluster.INSTANCE.start();
+ FlinkSpMiniCluster
+ .INSTANCE
+ .getClusterClient()
+
.submitJob(env.getStreamGraph(bindingParams.getGraph().getElementId()).getJobGraph());
+ }
} catch (Exception e) {
e.printStackTrace();
}
@@ -235,14 +244,14 @@ public abstract class FlinkRuntime<RP extends
RuntimeParams<B, I, RC>, B extends
@Override
public void discardRuntime() throws SpRuntimeException {
try {
- RestClusterClient<String> restClient = getRestClient();
+ ClusterClient<? extends Comparable<? extends Comparable<?>>>
clusterClient = getClusterClient();
Optional<JobStatusMessage> jobStatusMessage =
getJobStatus(bindingParams.getGraph().getElementId());
if (jobStatusMessage.isPresent()) {
String jobStatusStr = jobStatusMessage.get().getJobState().name();
// Cancel the job if running
if (jobStatusStr.equals("RUNNING")) {
- restClient.cancel(jobStatusMessage.get().getJobId());
+ clusterClient.cancel(jobStatusMessage.get().getJobId());
}
// else ignore, because job is already discarded
} else {
@@ -267,7 +276,19 @@ public abstract class FlinkRuntime<RP extends
RuntimeParams<B, I, RC>, B extends
}
}
- private RestClusterClient<String> getRestClient() throws Exception {
+ private ClusterClient<? extends Comparable<? extends Comparable<?>>>
getClusterClient() throws Exception {
+ if (config.isMiniClusterMode()) {
+ return getMiniClusterClient();
+ } else {
+ return getRestClusterClient();
+ }
+ }
+
+ private MiniClusterClient getMiniClusterClient() throws Exception {
+ return FlinkSpMiniCluster.INSTANCE.getClusterClient();
+ }
+
+ private RestClusterClient<String> getRestClusterClient() throws Exception {
Configuration restConfig = new Configuration();
restConfig.setString(JobManagerOptions.ADDRESS, config.getHost());
restConfig.setInteger(JobManagerOptions.PORT, config.getPort());
@@ -277,8 +298,8 @@ public abstract class FlinkRuntime<RP extends
RuntimeParams<B, I, RC>, B extends
private Optional<JobStatusMessage> getJobStatus(String jobName) {
try {
- RestClusterClient<String> restClient = getRestClient();
- CompletableFuture<Collection<JobStatusMessage>> jobs =
restClient.listJobs();
+ ClusterClient<? extends Comparable<? extends Comparable<?>>>
clusterClient = getClusterClient();
+ CompletableFuture<Collection<JobStatusMessage>> jobs =
clusterClient.listJobs();
Collection<JobStatusMessage> jobsFound = jobs.get();
// First, find a job with Running status
diff --git
a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkSpMiniCluster.java
b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkSpMiniCluster.java
new file mode 100644
index 0000000..8e773d2
--- /dev/null
+++
b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkSpMiniCluster.java
@@ -0,0 +1,38 @@
+package org.apache.streampipes.wrapper.flink;
+
+import org.apache.flink.client.program.MiniClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
+
+public enum FlinkSpMiniCluster {
+
+ INSTANCE;
+
+ private MiniCluster miniCluster;
+ private boolean miniClusterRunning;
+
+ FlinkSpMiniCluster() {
+ Configuration configuration = new Configuration();
+ configuration.setString(RestOptions.BIND_PORT, "0");
+ MiniClusterConfiguration miniClusterConfiguration = new
MiniClusterConfiguration
+ .Builder()
+ .setConfiguration(configuration)
+ .setNumTaskManagers(2)
+ .build();
+ this.miniCluster = new MiniCluster(miniClusterConfiguration);
+ }
+
+ public void start() throws Exception {
+ if (!this.miniClusterRunning) {
+ this.miniCluster.start();
+ this.miniClusterRunning = true;
+ }
+ }
+
+ public MiniClusterClient getClusterClient() {
+ return new MiniClusterClient(new Configuration(), miniCluster);
+ }
+
+}
