Skip to content

Commit 15c1919

Browse files
och5351och5351
committed
[Feature] [Flink] Define FlinkStreamingJob Java lifecycle base class
1 parent 4cd1140 commit 15c1919

6 files changed

Lines changed: 319 additions & 0 deletions

File tree

‎streampark-flink/pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
<name>StreamPark : Flink Parent</name>
3030

3131
<modules>
32+
<module>streampark-flink-core</module>
3233
<module>streampark-flink-shims</module>
3334
<module>streampark-flink-sqlclient</module>
3435
<module>streampark-flink-udf</module>

‎streampark-flink/streampark-flink-core/.flattened-pom.xml‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<!--
3+
~ Licensed to the Apache Software Foundation (ASF) under one or more
4+
~ contributor license agreements. See the NOTICE file distributed with
5+
~ this work for additional information regarding copyright ownership.
6+
~ The ASF licenses this file to You under the Apache License, Version 2.0
7+
~ (the "License"); you may not use this file except in compliance with
8+
~ the License. You may obtain a copy of the License at
9+
~
10+
~ http://www.apache.org/licenses/LICENSE-2.0
11+
~
12+
~ Unless required by applicable law or agreed to in writing, software
13+
~ distributed under the License is distributed on an "AS IS" BASIS,
14+
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
~ See the License for the specific language governing permissions and
16+
~ limitations under the License.
17+
-->
18+
<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
19+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
20+
<modelVersion>4.0.0</modelVersion>
21+
<parent>
22+
<groupId>org.apache.streampark</groupId>
23+
<artifactId>streampark-flink</artifactId>
24+
<version>3.0.0-SNAPSHOT</version>
25+
</parent>
26+
<artifactId>streampark-flink-core</artifactId>
27+
<version>3.0.0-SNAPSHOT</version>
28+
<name>StreamPark : Flink Core</name>
29+
<licenses>
30+
<license>
31+
<name>The Apache License, Version 2.0</name>
32+
<url>https://www.apache.org/licenses/LICENSE-2.0.txt</url>
33+
</license>
34+
</licenses>
35+
<dependencies>
36+
<dependency>
37+
<groupId>org.apache.flink</groupId>
38+
<artifactId>flink-streaming-java</artifactId>
39+
<version>${flink.version}</version>
40+
<scope>provided</scope>
41+
</dependency>
42+
<dependency>
43+
<groupId>org.apache.flink</groupId>
44+
<artifactId>flink-clients</artifactId>
45+
<version>${flink.version}</version>
46+
<scope>provided</scope>
47+
</dependency>
48+
<dependency>
49+
<groupId>org.junit.jupiter</groupId>
50+
<artifactId>junit-jupiter-engine</artifactId>
51+
<scope>test</scope>
52+
</dependency>
53+
<dependency>
54+
<groupId>org.assertj</groupId>
55+
<artifactId>assertj-core</artifactId>
56+
<scope>test</scope>
57+
</dependency>
58+
</dependencies>
59+
<profiles>
60+
<profile>
61+
<id>apache-release</id>
62+
<properties>
63+
<maven.deploy.skip>true</maven.deploy.skip>
64+
</properties>
65+
</profile>
66+
</profiles>
67+
</project>

‎streampark-flink/streampark-flink-core/pom.xml‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<!--
3+
~ Licensed to the Apache Software Foundation (ASF) under one or more
4+
~ contributor license agreements. See the NOTICE file distributed with
5+
~ this work for additional information regarding copyright ownership.
6+
~ The ASF licenses this file to You under the Apache License, Version 2.0
7+
~ (the "License"); you may not use this file except in compliance with
8+
~ the License. You may obtain a copy of the License at
9+
~
10+
~ http://www.apache.org/licenses/LICENSE-2.0
11+
~
12+
~ Unless required by applicable law or agreed to in writing, software
13+
~ distributed under the License is distributed on an "AS IS" BASIS,
14+
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
~ See the License for the specific language governing permissions and
16+
~ limitations under the License.
17+
-->
18+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
19+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
20+
<modelVersion>4.0.0</modelVersion>
21+
<parent>
22+
<groupId>org.apache.streampark</groupId>
23+
<artifactId>streampark-flink</artifactId>
24+
<version>${revision}</version>
25+
</parent>
26+
27+
<artifactId>streampark-flink-core</artifactId>
28+
<name>StreamPark : Flink Core</name>
29+
30+
<dependencies>
31+
<dependency>
32+
<groupId>org.apache.flink</groupId>
33+
<artifactId>flink-streaming-java</artifactId>
34+
<version>${flink.version}</version>
35+
<scope>provided</scope>
36+
</dependency>
37+
38+
<dependency>
39+
<groupId>org.apache.flink</groupId>
40+
<artifactId>flink-clients</artifactId>
41+
<version>${flink.version}</version>
42+
<scope>provided</scope>
43+
</dependency>
44+
45+
<dependency>
46+
<groupId>org.junit.jupiter</groupId>
47+
<artifactId>junit-jupiter-engine</artifactId>
48+
<scope>test</scope>
49+
</dependency>
50+
51+
<dependency>
52+
<groupId>org.assertj</groupId>
53+
<artifactId>assertj-core</artifactId>
54+
<scope>test</scope>
55+
</dependency>
56+
</dependencies>
57+
58+
<profiles>
59+
<profile>
60+
<id>apache-release</id>
61+
<properties>
62+
<maven.deploy.skip>true</maven.deploy.skip>
63+
</properties>
64+
</profile>
65+
</profiles>
66+
67+
</project>

‎streampark-flink/streampark-flink-core/src/main/java/org/apache/streampark/flink/core/FlinkJobException.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.streampark.flink.core;
19+
20+
/** Thrown when a {@link FlinkStreamingJob} fails during lifecycle execution. */
21+
public class FlinkJobException extends Exception {
22+
23+
public FlinkJobException(String message, Throwable cause) {
24+
super(message, cause);
25+
}
26+
}

‎streampark-flink/streampark-flink-core/src/main/java/org/apache/streampark/flink/core/FlinkStreamingJob.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.streampark.flink.core;
19+
20+
import org.apache.flink.api.common.JobExecutionResult;
21+
import org.apache.flink.api.java.utils.ParameterTool;
22+
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
23+
24+
public abstract class FlinkStreamingJob {
25+
26+
protected StreamExecutionEnvironment env;
27+
protected ParameterTool parameter;
28+
protected JobExecutionResult jobExecutionResult;
29+
30+
public final void run(String[] args) throws FlinkJobException {
31+
try {
32+
init(args);
33+
ready();
34+
handle();
35+
jobExecutionResult =
36+
env.execute(parameter.get("app.name", getClass().getSimpleName()));
37+
destroy();
38+
} catch (Exception e) {
39+
throw new FlinkJobException("Failed to run Flink job: " + getClass().getSimpleName(), e);
40+
}
41+
}
42+
43+
private void init(String[] args) {
44+
this.parameter = ParameterTool.fromArgs(args);
45+
this.env = StreamExecutionEnvironment.getExecutionEnvironment();
46+
configure(env, parameter);
47+
env.getConfig().setGlobalJobParameters(parameter);
48+
}
49+
50+
protected void configure(StreamExecutionEnvironment env, ParameterTool parameter) {
51+
}
52+
53+
protected void ready() {
54+
}
55+
56+
protected abstract void handle() throws FlinkJobException;
57+
58+
protected void destroy() {
59+
}
60+
}

‎streampark-flink/streampark-flink-core/src/test/java/org/apache/streampark/flink/core/FlinkStreamingJobTest.java‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.streampark.flink.core;
19+
20+
import org.apache.flink.api.java.utils.ParameterTool;
21+
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
22+
23+
import org.junit.jupiter.api.Test;
24+
25+
import java.util.ArrayList;
26+
import java.util.List;
27+
28+
import static org.assertj.core.api.Assertions.assertThat;
29+
30+
class FlinkStreamingJobTest {
31+
32+
@Test
33+
void lifecycleExecutesInOrder() throws Exception {
34+
List<String> trace = new ArrayList<>();
35+
36+
FlinkStreamingJob job =
37+
new FlinkStreamingJob() {
38+
39+
@Override
40+
protected void configure(StreamExecutionEnvironment env, ParameterTool parameter) {
41+
trace.add("configure");
42+
}
43+
44+
@Override
45+
protected void ready() {
46+
trace.add("ready");
47+
}
48+
49+
@Override
50+
protected void handle() {
51+
trace.add("handle");
52+
env.fromElements(1, 2, 3).map(i -> i * 2).print();
53+
}
54+
55+
@Override
56+
protected void destroy() {
57+
trace.add("destroy");
58+
}
59+
};
60+
61+
job.run(new String[0]);
62+
63+
assertThat(trace).containsExactly("configure", "ready", "handle", "destroy");
64+
assertThat(job.jobExecutionResult).isNotNull();
65+
}
66+
67+
static class NoOpJob extends FlinkStreamingJob {
68+
69+
@Override
70+
protected void handle() throws FlinkJobException {
71+
env.fromElements(1).print();
72+
}
73+
}
74+
75+
@Test
76+
void appNameDefaultsToClassName() throws Exception {
77+
NoOpJob job = new NoOpJob();
78+
job.run(new String[0]);
79+
assertThat(job.getClass().getSimpleName()).isEqualTo("NoOpJob");
80+
assertThat(job.jobExecutionResult).isNotNull();
81+
}
82+
83+
@Test
84+
void parameterToolPropagatedToEnv() throws Exception {
85+
FlinkStreamingJob job =
86+
new FlinkStreamingJob() {
87+
88+
@Override
89+
protected void handle() {
90+
env.fromElements(1).print();
91+
}
92+
};
93+
94+
job.run(new String[]{"--app.name", "test-job"});
95+
96+
assertThat(job.parameter.get("app.name")).isEqualTo("test-job");
97+
}
98+
}

0 commit comments

Comments
 (0)