diff --git a/streampark-flink/pom.xml b/streampark-flink/pom.xml index 83388a1207..098aa77c8e 100644 --- a/streampark-flink/pom.xml +++ b/streampark-flink/pom.xml @@ -29,6 +29,7 @@ StreamPark : Flink Parent + streampark-flink-core streampark-flink-shims streampark-flink-sqlclient streampark-flink-udf diff --git a/streampark-flink/streampark-flink-core/pom.xml b/streampark-flink/streampark-flink-core/pom.xml new file mode 100644 index 0000000000..7d9724db9c --- /dev/null +++ b/streampark-flink/streampark-flink-core/pom.xml @@ -0,0 +1,67 @@ + + + + 4.0.0 + + org.apache.streampark + streampark-flink + ${revision} + + + streampark-flink-core + StreamPark : Flink Core + + + + org.apache.flink + flink-streaming-java + ${flink.version} + provided + + + + org.apache.flink + flink-clients + ${flink.version} + provided + + + + org.junit.jupiter + junit-jupiter-engine + test + + + + org.assertj + assertj-core + test + + + + + + apache-release + + true + + + + + diff --git a/streampark-flink/streampark-flink-core/src/main/java/org/apache/streampark/flink/core/FlinkJobException.java b/streampark-flink/streampark-flink-core/src/main/java/org/apache/streampark/flink/core/FlinkJobException.java new file mode 100644 index 0000000000..82aadd0a65 --- /dev/null +++ b/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); + } +} diff --git a/streampark-flink/streampark-flink-core/src/main/java/org/apache/streampark/flink/core/FlinkStreamingJob.java b/streampark-flink/streampark-flink-core/src/main/java/org/apache/streampark/flink/core/FlinkStreamingJob.java new file mode 100644 index 0000000000..53026425bb --- /dev/null +++ b/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() { + } +} diff --git a/streampark-flink/streampark-flink-core/src/test/java/org/apache/streampark/flink/core/FlinkStreamingJobTest.java b/streampark-flink/streampark-flink-core/src/test/java/org/apache/streampark/flink/core/FlinkStreamingJobTest.java new file mode 100644 index 0000000000..817a63a90d --- /dev/null +++ b/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 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"); + } +}