[Feature][Flink] Define FlinkStreamingJob Java lifecycle base class

streampark-flink/pom.xml

@@ -29,6 +29,7 @@
     <name>StreamPark : Flink Parent</name>
 
     <modules>
+        <module>streampark-flink-core</module>
         <module>streampark-flink-shims</module>
         <module>streampark-flink-sqlclient</module>
         <module>streampark-flink-udf</module>

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

@@ -0,0 +1,67 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.streampark</groupId>
+        <artifactId>streampark-flink</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>streampark-flink-core</artifactId>
+    <name>StreamPark : Flink Core</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-clients</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <profiles>
+        <profile>
+            <id>apache-release</id>
+            <properties>
+                <maven.deploy.skip>true</maven.deploy.skip>
+            </properties>
+        </profile>
+    </profiles>
+
+</project>

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

@@ -0,0 +1,26 @@
+/*
+ * 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.streampark.flink.core;
+
+/** Thrown when a {@link FlinkStreamingJob} fails during lifecycle execution. */
+public class FlinkJobException extends Exception {
+
+    public FlinkJobException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

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

@@ -0,0 +1,60 @@
+/*
+ * 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.streampark.flink.core;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+public abstract class FlinkStreamingJob {
+
+    protected StreamExecutionEnvironment env;
+    protected ParameterTool parameter;
+    protected JobExecutionResult jobExecutionResult;
+
+    public final void run(String[] args) throws FlinkJobException {
+        try {
+            init(args);
+            ready();
+            handle();
+            jobExecutionResult =
+                env.execute(parameter.get("app.name", getClass().getSimpleName()));
+            destroy();
+        } catch (Exception e) {
+            throw new FlinkJobException("Failed to run Flink job: " + getClass().getSimpleName(), e);
+        }
+    }
+
+    private void init(String[] args) {
+        this.parameter = ParameterTool.fromArgs(args);
+        this.env = StreamExecutionEnvironment.getExecutionEnvironment();
+        configure(env, parameter);
+        env.getConfig().setGlobalJobParameters(parameter);
+    }
+
+    protected void configure(StreamExecutionEnvironment env, ParameterTool parameter) {
+    }
+
+    protected void ready() {
+    }
+
+    protected abstract void handle() throws FlinkJobException;
+
+    protected void destroy() {
+    }
+}

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

@@ -0,0 +1,98 @@
+/*
+ * 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.streampark.flink.core;
+
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class FlinkStreamingJobTest {
+
+    @Test
+    void lifecycleExecutesInOrder() throws Exception {
+        List<String> trace = new ArrayList<>();
+
+        FlinkStreamingJob job =
+            new FlinkStreamingJob() {
+
+                @Override
+                protected void configure(StreamExecutionEnvironment env, ParameterTool parameter) {
+                    trace.add("configure");
+                }
+
+                @Override
+                protected void ready() {
+                    trace.add("ready");
+                }
+
+                @Override
+                protected void handle() {
+                    trace.add("handle");
+                    env.fromElements(1, 2, 3).map(i -> i * 2).print();
+                }
+
+                @Override
+                protected void destroy() {
+                    trace.add("destroy");
+                }
+            };
+
+        job.run(new String[0]);
+
+        assertThat(trace).containsExactly("configure", "ready", "handle", "destroy");
+        assertThat(job.jobExecutionResult).isNotNull();
+    }
+
+    static class NoOpJob extends FlinkStreamingJob {
+
+        @Override
+        protected void handle() throws FlinkJobException {
+            env.fromElements(1).print();
+        }
+    }
+
+    @Test
+    void appNameDefaultsToClassName() throws Exception {
+        NoOpJob job = new NoOpJob();
+        job.run(new String[0]);
+        assertThat(job.getClass().getSimpleName()).isEqualTo("NoOpJob");
+        assertThat(job.jobExecutionResult).isNotNull();
+    }
+
+    @Test
+    void parameterToolPropagatedToEnv() throws Exception {
+        FlinkStreamingJob job =
+            new FlinkStreamingJob() {
+
+                @Override
+                protected void handle() {
+                    env.fromElements(1).print();
+                }
+            };
+
+        job.run(new String[]{"--app.name", "test-job"});
+
+        assertThat(job.parameter.get("app.name")).isEqualTo("test-job");
+    }
+}