From ef1716e87944650d1c8f4a26b9c51375a85ee958 Mon Sep 17 00:00:00 2001 From: ocb3916 Date: Tue, 21 Jul 2026 12:36:59 +0900 Subject: [PATCH 1/4] [Flink] Define FlinkTableJob and FlinkStreamTableJob Java base classes - Add Java lifecycle base classes for Table API and Streaming+Table API jobs, following the same ready/handle/destroy lifecycle established by FlinkStreamingJob (sub-issue 0.1). - Use composition (getTableEnv()/getEnv()) instead of extending/mimicking TableEnvironment / StreamTableEnvironment, avoiding ~40-50 delegate methods per class that the legacy Scala traits needed. - Add JUnit 5 smoke tests covering lifecycle order, SQL bridging, and the markConvertedToDataStream() flag. Part of #4408 (Flink Scala-to-Java migration), Phase 0.2. --- .../core/javaapi/FlinkStreamTableJob.java | 164 ++++++++++++++++++ .../flink/core/javaapi/FlinkTableJob.java | 151 ++++++++++++++++ .../core/javaapi/FlinkStreamTableJobTest.java | 88 ++++++++++ .../flink/core/javaapi/FlinkTableJobTest.java | 121 +++++++++++++ 4 files changed, 524 insertions(+) create mode 100644 streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java create mode 100644 streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java create mode 100644 streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJobTest.java create mode 100644 streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java new file mode 100644 index 0000000000..4a0639fe5e --- /dev/null +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java @@ -0,0 +1,164 @@ +/* + * 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.javaapi; + +import org.apache.streampark.flink.core.FlinkSqlExecutor$; + +import org.apache.flink.api.common.JobExecutionResult; +import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; + +import java.util.function.Consumer; + +import scala.Function1; +import scala.runtime.AbstractFunction1; +import scala.runtime.BoxedUnit; + +/** + * Java lifecycle base class for jobs that mix the DataStream API and the Table API. + * + *

Unlike the legacy Scala {@code FlinkStreamTableTrait}, this class does NOT extend / mimic + * {@link StreamTableEnvironment} or {@link StreamExecutionEnvironment}. Instead it holds both by + * composition — call {@link #getEnv()} / {@link #getTableEnv()} directly instead of relying on + * delegate methods (the legacy trait needed a {@code $}-prefix naming hack to avoid clashes + * between the two APIs; composition removes that need entirely). + * + *

Note on {@code toDataStream} conversions: the legacy trait auto-detected when a + * {@code Table} was converted back to a {@code DataStream} (via an override) so {@code start()} + * knew whether to also call {@code env.execute()}. Because this class no longer intercepts calls + * made directly on {@link #getTableEnv()}, that auto-detection is not possible — call {@link + * #markConvertedToDataStream()} explicitly after such a conversion in {@link #handle()}. + */ +public abstract class FlinkStreamTableJob { + + protected final ParameterTool parameter; + + protected final StreamExecutionEnvironment env; + + protected final StreamTableEnvironment tableEnv; + + private boolean convertedToDataStream = false; + + protected FlinkStreamTableJob( + ParameterTool parameter, StreamExecutionEnvironment env, + StreamTableEnvironment tableEnv) { + this.parameter = parameter; + this.env = env; + this.tableEnv = tableEnv; + } + + /** Recommended entry point to start the job. */ + public final JobExecutionResult start() throws Exception { + ready(); + handle(); + JobExecutionResult result = execute(getAppName()); + destroy(); + return result; + } + + /** Hook called before {@link #handle()}. Override for pre-job setup. */ + protected void ready() { + } + + /** User job logic goes here — build the DataStream / Table pipeline. */ + protected abstract void handle() throws Exception; + + /** Hook called after {@link #execute(String)}. Override for cleanup. */ + protected void destroy() { + } + + /** + * Executes the job under the given name. Only triggers {@link + * StreamExecutionEnvironment#execute(String)} if the pipeline was converted back to a {@code + * DataStream} (see {@link #markConvertedToDataStream()}); pure Table/SQL pipelines are already + * triggered inside {@link #handle()} via {@code executeSql} / {@code StatementSet#execute()}. + */ + protected JobExecutionResult execute(String jobName) throws Exception { + // TODO: replace with Utils.printLogo(...) once streampark-common Java migration (Phase 1) lands + System.out.println("[StreamPark] FlinkStreamTable " + jobName + " Starting..."); + if (convertedToDataStream) { + return env.execute(jobName); + } + return null; + } + + /** + * Convenience shortcut for running a single SQL statement, matching legacy {@code sql(...)}. + * Statement result lines (e.g. from {@code SHOW TABLES}, {@code EXPLAIN}) are logged by + * default — use {@link #sql(String, Consumer)} to receive them instead. + */ + public void sql(String sql) { + sql(sql, null); + } + + /** + * Runs a single SQL statement, routing any result lines to the given callback instead of the + * default log output. + * + *

{@code FlinkSqlExecutor.executeSql} (not yet migrated off Scala — see Phase 3) takes a + * Scala {@code String => Unit} as its 4th argument; this bridges a plain Java {@link Consumer} + * to that type so the public surface of this class stays Scala-free. + */ + public void sql(String sql, Consumer callback) { + Function1 scalaCallback = + callback == null + ? null + : new AbstractFunction1() { + + @Override + public BoxedUnit apply(String result) { + callback.accept(result); + return BoxedUnit.UNIT; + } + }; + FlinkSqlExecutor$.MODULE$.executeSql(sql, parameter, tableEnv, scalaCallback); + } + + /** + * Call this after converting a {@code Table} back to a {@code DataStream} (e.g. via {@code + * getTableEnv().toDataStream(table)}) so that {@link #execute(String)} knows to also run the + * underlying {@link StreamExecutionEnvironment}. + */ + protected void markConvertedToDataStream() { + this.convertedToDataStream = true; + } + + /** Direct access to the DataStream API — use this instead of {@code $}-prefixed delegates. */ + public StreamExecutionEnvironment getEnv() { + return env; + } + + /** Direct access to the Table API — use this instead of delegate methods. */ + public StreamTableEnvironment getTableEnv() { + return tableEnv; + } + + public ParameterTool getParameter() { + return parameter; + } + + private String getAppName() { + // TODO: mirror EnhancerImplicit#getAppName(required = true) once ported to Java (Phase 1.3) + String appName = parameter.get("app.name"); + if (appName == null || appName.isEmpty()) { + throw new IllegalArgumentException("[StreamPark] \"app.name\" is required"); + } + return appName; + } +} diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java new file mode 100644 index 0000000000..d7e833c634 --- /dev/null +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java @@ -0,0 +1,151 @@ +/* + * 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.javaapi; + +import org.apache.streampark.flink.core.FlinkSqlExecutor$; + +import org.apache.flink.api.common.JobExecutionResult; +import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.table.api.TableEnvironment; + +import java.util.function.Consumer; + +import scala.Function1; +import scala.runtime.AbstractFunction1; +import scala.runtime.BoxedUnit; + +/** + * Java lifecycle base class for Flink Table API (batch) jobs. + * + *

Unlike the legacy Scala {@code FlinkTableTrait}, this class does NOT extend / mimic {@link + * TableEnvironment}. Instead it holds a {@link TableEnvironment} by composition — call {@link + * #getTableEnv()} to access the full Table API surface directly. This avoids re-implementing and + * maintaining dozens of delegate methods every time the Flink Table API changes. + * + *

Typical usage: + * + * 

{@code
+ * public class MyTableJob extends FlinkTableJob {
+ *
+ *   public MyTableJob(ParameterTool parameter, TableEnvironment tableEnv) {
+ *     super(parameter, tableEnv);
+ *   }
+ *
+ *   @Override
+ *   protected void handle() throws Exception {
+ *     getTableEnv().executeSql("SELECT * FROM my_table");
+ *   }
+ * }
+ * }
+ */ +public abstract class FlinkTableJob { + + protected final ParameterTool parameter; + + protected final TableEnvironment tableEnv; + + protected FlinkTableJob(ParameterTool parameter, TableEnvironment tableEnv) { + this.parameter = parameter; + this.tableEnv = tableEnv; + } + + /** + * Recommended entry point to start the job. Mirrors the legacy trait's {@code start()}: runs + * the fixed lifecycle and executes under the required {@code app.name} parameter. + */ + public final JobExecutionResult start() throws Exception { + ready(); + handle(); + JobExecutionResult result = execute(getAppName()); + destroy(); + return result; + } + + /** Hook called before {@link #handle()}. Override for pre-job setup (e.g. catalogs, UDFs). */ + protected void ready() { + } + + /** User job logic goes here — build and register the Table API pipeline. */ + protected abstract void handle() throws Exception; + + /** Hook called after {@link #execute(String)}. Override for cleanup. */ + protected void destroy() { + } + + /** + * Executes the job under the given name. Batch Table API jobs are typically already triggered + * synchronously by {@code executeSql} / {@code StatementSet#execute()} inside {@link + * #handle()}, so this returns {@code null} by default — override if a specific job result is + * needed. + */ + protected JobExecutionResult execute(String jobName) throws Exception { + // TODO: replace with Utils.printLogo(...) once streampark-common Java migration (Phase 1) lands + System.out.println("[StreamPark] FlinkTable " + jobName + " Starting..."); + return null; + } + + /** + * Convenience shortcut for running a single SQL statement, matching legacy {@code sql(...)}. + * Statement result lines (e.g. from {@code SHOW TABLES}, {@code EXPLAIN}) are logged by + * default — use {@link #sql(String, Consumer)} to receive them instead. + */ + public void sql(String sql) { + sql(sql, null); + } + + /** + * Runs a single SQL statement, routing any result lines to the given callback instead of the + * default log output. + * + *

{@code FlinkSqlExecutor.executeSql} (not yet migrated off Scala — see Phase 3) takes a + * Scala {@code String => Unit} as its 4th argument; this bridges a plain Java {@link Consumer} + * to that type so the public surface of this class stays Scala-free. + */ + public void sql(String sql, Consumer callback) { + Function1 scalaCallback = + callback == null + ? null + : new AbstractFunction1() { + + @Override + public BoxedUnit apply(String result) { + callback.accept(result); + return BoxedUnit.UNIT; + } + }; + FlinkSqlExecutor$.MODULE$.executeSql(sql, parameter, tableEnv, scalaCallback); + } + + /** Direct access to the underlying Table API — use this instead of delegate methods. */ + public TableEnvironment getTableEnv() { + return tableEnv; + } + + public ParameterTool getParameter() { + return parameter; + } + + private String getAppName() { + // TODO: mirror EnhancerImplicit#getAppName(required = true) once ported to Java (Phase 1.3) + String appName = parameter.get("app.name"); + if (appName == null || appName.isEmpty()) { + throw new IllegalArgumentException("[StreamPark] \"app.name\" is required"); + } + return appName; + } +} diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJobTest.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJobTest.java new file mode 100644 index 0000000000..f6f7cb8f02 --- /dev/null +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJobTest.java @@ -0,0 +1,88 @@ +/* + * 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.javaapi; + +import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.types.Row; + +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertSame; + +/** Smoke tests for {@link FlinkStreamTableJob} — verifies composition and the dataStream flag. */ +class FlinkStreamTableJobTest { + + private static ParameterTool paramWithAppName(String appName) { + Map map = new HashMap<>(); + map.put("app.name", appName); + return ParameterTool.fromMap(map); + } + + @Test + void handleGetsDirectAccessToEnvAndTableEnv() throws Exception { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tableEnv = + StreamTableEnvironment.create(env, EnvironmentSettings.newInstance().inStreamingMode().build()); + + FlinkStreamTableJob job = + new FlinkStreamTableJob(paramWithAppName("test-stream-table-job"), env, tableEnv) { + + @Override + protected void handle() { + assertSame(env, getEnv()); + assertSame(tableEnv, getTableEnv()); + } + }; + + job.start(); + } + + @Test + void markConvertedToDataStreamTriggersEnvExecute() { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tableEnv = + StreamTableEnvironment.create(env, EnvironmentSettings.newInstance().inStreamingMode().build()); + + FlinkStreamTableJob job = + new FlinkStreamTableJob(paramWithAppName("test-stream-table-job"), env, tableEnv) { + + @Override + protected void handle() { + Table table = getTableEnv().fromValues("a", "b", "c"); + DataStream stream = getTableEnv().toDataStream(table); + stream.addSink(new SinkFunction() { + }); // no-op sink so the graph has something to execute + markConvertedToDataStream(); + } + }; + + // If markConvertedToDataStream() didn't take effect, env.execute() would never run and + // this bounded stream job would just silently return null instead of actually executing. + assertDoesNotThrow(job::start); + } +} diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java new file mode 100644 index 0000000000..3ada9033ff --- /dev/null +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java @@ -0,0 +1,121 @@ +/* + * 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.javaapi; + +import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** Smoke tests for {@link FlinkTableJob} — verifies the lifecycle contract and SQL bridging. */ +class FlinkTableJobTest { + + private static ParameterTool paramWithAppName(String appName) { + Map map = new HashMap<>(); + map.put("app.name", appName); + return ParameterTool.fromMap(map); + } + + private static TableEnvironment newBatchTableEnv() { + return TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build()); + } + + @Test + void lifecycleRunsInOrderAndExposesTableEnv() throws Exception { + List calls = new ArrayList<>(); + TableEnvironment tableEnv = newBatchTableEnv(); + + FlinkTableJob job = + new FlinkTableJob(paramWithAppName("test-table-job"), tableEnv) { + + @Override + protected void ready() { + calls.add("ready"); + } + + @Override + protected void handle() { + calls.add("handle"); + // getTableEnv() must return exactly the instance we constructed with — + // proves composition, not a copy or a re-wrapped object. + assertSame(tableEnv, getTableEnv()); + } + + @Override + protected void destroy() { + calls.add("destroy"); + } + }; + + job.start(); + + assertEquals(List.of("ready", "handle", "destroy"), calls); + } + + @Test + void sqlRunsAgainstRealTableEnvAndInvokesCallback() { + TableEnvironment tableEnv = newBatchTableEnv(); + tableEnv.executeSql("CREATE TABLE sink (msg STRING) WITH ('connector' = 'print')"); + + // FlinkSqlExecutor.executeSql treats a non-blank `sql` argument as a *parameter lookup + // key*, not literal SQL text — the actual statement must be registered under that key in + // the ParameterTool first (this mirrors how it's driven from job config in practice). + Map params = new HashMap<>(); + params.put("app.name", "test-table-job"); + params.put("my.insert.sql", "INSERT INTO sink VALUES ('hello')"); + ParameterTool parameter = ParameterTool.fromMap(params); + + List callbackLines = new ArrayList<>(); + FlinkTableJob job = + new FlinkTableJob(parameter, tableEnv) { + + @Override + protected void handle() { + sql("my.insert.sql", callbackLines::add); + } + }; + + assertDoesNotThrow(job::start); + } + + @Test + void missingAppNameThrowsBeforeExecuting() { + TableEnvironment tableEnv = newBatchTableEnv(); + Map empty = new HashMap<>(); + FlinkTableJob job = + new FlinkTableJob(ParameterTool.fromMap(empty), tableEnv) { + + @Override + protected void handle() { + } + }; + + assertThrows(IllegalArgumentException.class, job::start); + } +} From dfb0d31cd9a697b71450ec53d61462283185b28c Mon Sep 17 00:00:00 2001 From: ocb3916 Date: Tue, 21 Jul 2026 15:34:16 +0900 Subject: [PATCH 2/4] [Flink] Define FlinkTableJob and FlinkStreamTableJob Java base classes - Add Java lifecycle base classes for Table API and Streaming+Table API jobs, following the same ready/handle/destroy lifecycle established by FlinkStreamingJob (sub-issue 0.1). - Use composition (getTableEnv()/getEnv()) instead of extending/mimicking TableEnvironment / StreamTableEnvironment, avoiding ~40-50 delegate methods per class that the legacy Scala traits needed. - Add JUnit 5 smoke tests covering lifecycle order, SQL bridging, and the markConvertedToDataStream() flag. Part of #4408 (Flink Scala-to-Java migration), Phase 0.2. --- .../apache/streampark/flink/core/javaapi/FlinkTableJobTest.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java index 3ada9033ff..fa9eb28e0d 100644 --- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java @@ -113,6 +113,8 @@ void missingAppNameThrowsBeforeExecuting() { @Override protected void handle() { + // Intentionally empty — start() must throw on the missing "app.name" before + // handle() is ever reached, so this body is never executed. } }; From b8395f88135eb0efe2218f4dad363961c61ddb90 Mon Sep 17 00:00:00 2001 From: ocb3916 Date: Tue, 21 Jul 2026 15:44:07 +0900 Subject: [PATCH 3/4] [Flink] Fix SonarCloud code duplication in Job base classes --- .../flink/core/javaapi/FlinkJobSupport.java | 80 +++++++++++++++++++ .../core/javaapi/FlinkStreamTableJob.java | 26 +----- .../flink/core/javaapi/FlinkTableJob.java | 26 +----- .../core/javaapi/FlinkStreamTableJobTest.java | 14 +--- .../flink/core/javaapi/FlinkTableJobTest.java | 8 +- .../flink/core/javaapi/JobTestParams.java | 36 +++++++++ 6 files changed, 123 insertions(+), 67 deletions(-) create mode 100644 streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkJobSupport.java create mode 100644 streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/JobTestParams.java diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkJobSupport.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkJobSupport.java new file mode 100644 index 0000000000..9a9d6eb1fb --- /dev/null +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkJobSupport.java @@ -0,0 +1,80 @@ +/* + * 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.javaapi; + +import org.apache.streampark.flink.core.FlinkSqlExecutor$; + +import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.table.api.TableEnvironment; + +import java.util.function.Consumer; + +import scala.Function1; +import scala.runtime.AbstractFunction1; +import scala.runtime.BoxedUnit; + +/** + * Shared, package-private helpers for {@link FlinkTableJob} and {@link FlinkStreamTableJob}. + * Pulled out to avoid duplicating the {@code app.name} lookup and the {@code FlinkSqlExecutor} + * Java/Scala callback bridge across both classes. + */ +final class FlinkJobSupport { + + private FlinkJobSupport() { + } + + /** + * Runs a single SQL statement via the (still-Scala) {@code FlinkSqlExecutor}, bridging a + * plain Java {@link Consumer} to the Scala {@code String => Unit} callback it expects so + * neither caller needs to touch Scala types directly. + * + *

{@code FlinkSqlExecutor.executeSql} is {@code private[streampark]}, so Scala does not + * generate a public static forwarder for it — it must be called via the module instance + * ({@code FlinkSqlExecutor$.MODULE$}) rather than {@code FlinkSqlExecutor.executeSql(...)}. + */ + static void executeSql( + String sql, ParameterTool parameter, TableEnvironment tableEnv, + Consumer callback) { + Function1 scalaCallback = + callback == null + ? null + : new AbstractFunction1() { + + @Override + public BoxedUnit apply(String result) { + callback.accept(result); + return BoxedUnit.UNIT; + } + }; + FlinkSqlExecutor$.MODULE$.executeSql(sql, parameter, tableEnv, scalaCallback); + } + + /** + * Returns the required {@code app.name} parameter, or throws if it's missing. + * + *

TODO: mirror {@code EnhancerImplicit#getAppName(required = true)} once ported to Java + * (Phase 1.3), and delegate to it from here instead. + */ + static String requireAppName(ParameterTool parameter) { + String appName = parameter.get("app.name"); + if (appName == null || appName.isEmpty()) { + throw new IllegalArgumentException("[StreamPark] \"app.name\" is required"); + } + return appName; + } +} diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java index 4a0639fe5e..e4ce14117f 100644 --- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java @@ -17,8 +17,6 @@ package org.apache.streampark.flink.core.javaapi; -import org.apache.streampark.flink.core.FlinkSqlExecutor$; - import org.apache.flink.api.common.JobExecutionResult; import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; @@ -26,10 +24,6 @@ import java.util.function.Consumer; -import scala.Function1; -import scala.runtime.AbstractFunction1; -import scala.runtime.BoxedUnit; - /** * Java lifecycle base class for jobs that mix the DataStream API and the Table API. * @@ -116,18 +110,7 @@ public void sql(String sql) { * to that type so the public surface of this class stays Scala-free. */ public void sql(String sql, Consumer callback) { - Function1 scalaCallback = - callback == null - ? null - : new AbstractFunction1() { - - @Override - public BoxedUnit apply(String result) { - callback.accept(result); - return BoxedUnit.UNIT; - } - }; - FlinkSqlExecutor$.MODULE$.executeSql(sql, parameter, tableEnv, scalaCallback); + FlinkJobSupport.executeSql(sql, parameter, tableEnv, callback); } /** @@ -154,11 +137,6 @@ public ParameterTool getParameter() { } private String getAppName() { - // TODO: mirror EnhancerImplicit#getAppName(required = true) once ported to Java (Phase 1.3) - String appName = parameter.get("app.name"); - if (appName == null || appName.isEmpty()) { - throw new IllegalArgumentException("[StreamPark] \"app.name\" is required"); - } - return appName; + return FlinkJobSupport.requireAppName(parameter); } } diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java index d7e833c634..a2ddd43a19 100644 --- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java @@ -17,18 +17,12 @@ package org.apache.streampark.flink.core.javaapi; -import org.apache.streampark.flink.core.FlinkSqlExecutor$; - import org.apache.flink.api.common.JobExecutionResult; import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.table.api.TableEnvironment; import java.util.function.Consumer; -import scala.Function1; -import scala.runtime.AbstractFunction1; -import scala.runtime.BoxedUnit; - /** * Java lifecycle base class for Flink Table API (batch) jobs. * @@ -117,18 +111,7 @@ public void sql(String sql) { * to that type so the public surface of this class stays Scala-free. */ public void sql(String sql, Consumer callback) { - Function1 scalaCallback = - callback == null - ? null - : new AbstractFunction1() { - - @Override - public BoxedUnit apply(String result) { - callback.accept(result); - return BoxedUnit.UNIT; - } - }; - FlinkSqlExecutor$.MODULE$.executeSql(sql, parameter, tableEnv, scalaCallback); + FlinkJobSupport.executeSql(sql, parameter, tableEnv, callback); } /** Direct access to the underlying Table API — use this instead of delegate methods. */ @@ -141,11 +124,6 @@ public ParameterTool getParameter() { } private String getAppName() { - // TODO: mirror EnhancerImplicit#getAppName(required = true) once ported to Java (Phase 1.3) - String appName = parameter.get("app.name"); - if (appName == null || appName.isEmpty()) { - throw new IllegalArgumentException("[StreamPark] \"app.name\" is required"); - } - return appName; + return FlinkJobSupport.requireAppName(parameter); } } diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJobTest.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJobTest.java index f6f7cb8f02..c555b27872 100644 --- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJobTest.java +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJobTest.java @@ -17,7 +17,6 @@ package org.apache.streampark.flink.core.javaapi; -import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.sink.SinkFunction; @@ -28,21 +27,12 @@ import org.junit.jupiter.api.Test; -import java.util.HashMap; -import java.util.Map; - import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertSame; /** Smoke tests for {@link FlinkStreamTableJob} — verifies composition and the dataStream flag. */ class FlinkStreamTableJobTest { - private static ParameterTool paramWithAppName(String appName) { - Map map = new HashMap<>(); - map.put("app.name", appName); - return ParameterTool.fromMap(map); - } - @Test void handleGetsDirectAccessToEnvAndTableEnv() throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -50,7 +40,7 @@ void handleGetsDirectAccessToEnvAndTableEnv() throws Exception { StreamTableEnvironment.create(env, EnvironmentSettings.newInstance().inStreamingMode().build()); FlinkStreamTableJob job = - new FlinkStreamTableJob(paramWithAppName("test-stream-table-job"), env, tableEnv) { + new FlinkStreamTableJob(JobTestParams.withAppName("test-stream-table-job"), env, tableEnv) { @Override protected void handle() { @@ -69,7 +59,7 @@ void markConvertedToDataStreamTriggersEnvExecute() { StreamTableEnvironment.create(env, EnvironmentSettings.newInstance().inStreamingMode().build()); FlinkStreamTableJob job = - new FlinkStreamTableJob(paramWithAppName("test-stream-table-job"), env, tableEnv) { + new FlinkStreamTableJob(JobTestParams.withAppName("test-stream-table-job"), env, tableEnv) { @Override protected void handle() { diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java index fa9eb28e0d..4bc4363cf6 100644 --- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/FlinkTableJobTest.java @@ -36,12 +36,6 @@ /** Smoke tests for {@link FlinkTableJob} — verifies the lifecycle contract and SQL bridging. */ class FlinkTableJobTest { - private static ParameterTool paramWithAppName(String appName) { - Map map = new HashMap<>(); - map.put("app.name", appName); - return ParameterTool.fromMap(map); - } - private static TableEnvironment newBatchTableEnv() { return TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build()); } @@ -52,7 +46,7 @@ void lifecycleRunsInOrderAndExposesTableEnv() throws Exception { TableEnvironment tableEnv = newBatchTableEnv(); FlinkTableJob job = - new FlinkTableJob(paramWithAppName("test-table-job"), tableEnv) { + new FlinkTableJob(JobTestParams.withAppName("test-table-job"), tableEnv) { @Override protected void ready() { diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/JobTestParams.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/JobTestParams.java new file mode 100644 index 0000000000..2284f5f2b3 --- /dev/null +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/test/java/org/apache/streampark/flink/core/javaapi/JobTestParams.java @@ -0,0 +1,36 @@ +/* + * 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.javaapi; + +import org.apache.flink.api.java.utils.ParameterTool; + +import java.util.HashMap; +import java.util.Map; + +/** Shared test-only helper for building a minimal {@link ParameterTool} with an app name. */ +final class JobTestParams { + + private JobTestParams() { + } + + static ParameterTool withAppName(String appName) { + Map map = new HashMap<>(); + map.put("app.name", appName); + return ParameterTool.fromMap(map); + } +} From 4682c69898f7034c88523ce8d4eb689938ee79f4 Mon Sep 17 00:00:00 2001 From: ocb3916 Date: Tue, 21 Jul 2026 15:54:30 +0900 Subject: [PATCH 4/4] [Flink] Extract shared lifecycle into AbstractFlinkJob to fix remaining duplication --- .../flink/core/javaapi/AbstractFlinkJob.java | 89 +++++++++++++++++++ .../core/javaapi/FlinkStreamTableJob.java | 69 +++----------- .../flink/core/javaapi/FlinkTableJob.java | 72 +++------------ 3 files changed, 109 insertions(+), 121 deletions(-) create mode 100644 streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/AbstractFlinkJob.java diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/AbstractFlinkJob.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/AbstractFlinkJob.java new file mode 100644 index 0000000000..451e9be5fe --- /dev/null +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/AbstractFlinkJob.java @@ -0,0 +1,89 @@ +/* + * 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.javaapi; + +import org.apache.flink.api.common.JobExecutionResult; +import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.table.api.TableEnvironment; + +import java.util.function.Consumer; + +/** + * Package-private shared base for {@link FlinkTableJob} and {@link FlinkStreamTableJob} — holds + * the common {@code ready → handle → execute → destroy} lifecycle, the SQL bridge, and the + * {@code app.name} lookup, so neither subclass has to repeat them. + * + *

Not part of the public API: callers only ever see {@link FlinkTableJob} or {@link + * FlinkStreamTableJob}, never this type directly. + */ +abstract class AbstractFlinkJob { + + protected final ParameterTool parameter; + + protected final TableEnvironment tableEnv; + + protected AbstractFlinkJob(ParameterTool parameter, TableEnvironment tableEnv) { + this.parameter = parameter; + this.tableEnv = tableEnv; + } + + /** Recommended entry point to start the job. */ + public final JobExecutionResult start() throws Exception { + ready(); + handle(); + JobExecutionResult result = execute(FlinkJobSupport.requireAppName(parameter)); + destroy(); + return result; + } + + /** Hook called before {@link #handle()}. Override for pre-job setup. */ + protected void ready() { + } + + /** User job logic goes here — build and register the pipeline. */ + protected abstract void handle() throws Exception; + + /** Runs the job under the given name; subclasses decide what "running" means for them. */ + protected abstract JobExecutionResult execute(String jobName) throws Exception; + + /** Hook called after {@link #execute(String)}. Override for cleanup. */ + protected void destroy() { + } + + /** + * Convenience shortcut for running a single SQL statement, matching legacy {@code sql(...)}. + * Statement result lines (e.g. from {@code SHOW TABLES}, {@code EXPLAIN}) are logged by + * default — use {@link #sql(String, Consumer)} to receive them instead. + */ + public void sql(String sql) { + sql(sql, null); + } + + /** + * Runs a single SQL statement, routing any result lines to the given callback instead of the + * default log output. See {@link FlinkJobSupport#executeSql} for details on how this bridges + * to the (still-Scala) {@code FlinkSqlExecutor}. + */ + public void sql(String sql, Consumer callback) { + FlinkJobSupport.executeSql(sql, parameter, tableEnv, callback); + } + + public ParameterTool getParameter() { + return parameter; + } +} diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java index e4ce14117f..60f5677ab4 100644 --- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java @@ -22,7 +22,8 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; -import java.util.function.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Java lifecycle base class for jobs that mix the DataStream API and the Table API. @@ -39,42 +40,19 @@ * made directly on {@link #getTableEnv()}, that auto-detection is not possible — call {@link * #markConvertedToDataStream()} explicitly after such a conversion in {@link #handle()}. */ -public abstract class FlinkStreamTableJob { +public abstract class FlinkStreamTableJob extends AbstractFlinkJob { - protected final ParameterTool parameter; + private static final Logger LOG = LoggerFactory.getLogger(FlinkStreamTableJob.class); protected final StreamExecutionEnvironment env; - protected final StreamTableEnvironment tableEnv; - private boolean convertedToDataStream = false; protected FlinkStreamTableJob( ParameterTool parameter, StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) { - this.parameter = parameter; + super(parameter, tableEnv); this.env = env; - this.tableEnv = tableEnv; - } - - /** Recommended entry point to start the job. */ - public final JobExecutionResult start() throws Exception { - ready(); - handle(); - JobExecutionResult result = execute(getAppName()); - destroy(); - return result; - } - - /** Hook called before {@link #handle()}. Override for pre-job setup. */ - protected void ready() { - } - - /** User job logic goes here — build the DataStream / Table pipeline. */ - protected abstract void handle() throws Exception; - - /** Hook called after {@link #execute(String)}. Override for cleanup. */ - protected void destroy() { } /** @@ -83,36 +61,17 @@ protected void destroy() { * DataStream} (see {@link #markConvertedToDataStream()}); pure Table/SQL pipelines are already * triggered inside {@link #handle()} via {@code executeSql} / {@code StatementSet#execute()}. */ + @Override protected JobExecutionResult execute(String jobName) throws Exception { - // TODO: replace with Utils.printLogo(...) once streampark-common Java migration (Phase 1) lands - System.out.println("[StreamPark] FlinkStreamTable " + jobName + " Starting..."); + // TODO(#4408): replace with Utils.printLogo(...) once streampark-common Java migration + // (Phase 1) lands + LOG.info("[StreamPark] FlinkStreamTable {} Starting...", jobName); if (convertedToDataStream) { return env.execute(jobName); } return null; } - /** - * Convenience shortcut for running a single SQL statement, matching legacy {@code sql(...)}. - * Statement result lines (e.g. from {@code SHOW TABLES}, {@code EXPLAIN}) are logged by - * default — use {@link #sql(String, Consumer)} to receive them instead. - */ - public void sql(String sql) { - sql(sql, null); - } - - /** - * Runs a single SQL statement, routing any result lines to the given callback instead of the - * default log output. - * - *

{@code FlinkSqlExecutor.executeSql} (not yet migrated off Scala — see Phase 3) takes a - * Scala {@code String => Unit} as its 4th argument; this bridges a plain Java {@link Consumer} - * to that type so the public surface of this class stays Scala-free. - */ - public void sql(String sql, Consumer callback) { - FlinkJobSupport.executeSql(sql, parameter, tableEnv, callback); - } - /** * Call this after converting a {@code Table} back to a {@code DataStream} (e.g. via {@code * getTableEnv().toDataStream(table)}) so that {@link #execute(String)} knows to also run the @@ -129,14 +88,6 @@ public StreamExecutionEnvironment getEnv() { /** Direct access to the Table API — use this instead of delegate methods. */ public StreamTableEnvironment getTableEnv() { - return tableEnv; - } - - public ParameterTool getParameter() { - return parameter; - } - - private String getAppName() { - return FlinkJobSupport.requireAppName(parameter); + return (StreamTableEnvironment) tableEnv; } } diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java index a2ddd43a19..d87fe2f3be 100644 --- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java +++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java @@ -21,7 +21,8 @@ import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.table.api.TableEnvironment; -import java.util.function.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Java lifecycle base class for Flink Table API (batch) jobs. @@ -47,38 +48,12 @@ * } * } */ -public abstract class FlinkTableJob { +public abstract class FlinkTableJob extends AbstractFlinkJob { - protected final ParameterTool parameter; - - protected final TableEnvironment tableEnv; + private static final Logger LOG = LoggerFactory.getLogger(FlinkTableJob.class); protected FlinkTableJob(ParameterTool parameter, TableEnvironment tableEnv) { - this.parameter = parameter; - this.tableEnv = tableEnv; - } - - /** - * Recommended entry point to start the job. Mirrors the legacy trait's {@code start()}: runs - * the fixed lifecycle and executes under the required {@code app.name} parameter. - */ - public final JobExecutionResult start() throws Exception { - ready(); - handle(); - JobExecutionResult result = execute(getAppName()); - destroy(); - return result; - } - - /** Hook called before {@link #handle()}. Override for pre-job setup (e.g. catalogs, UDFs). */ - protected void ready() { - } - - /** User job logic goes here — build and register the Table API pipeline. */ - protected abstract void handle() throws Exception; - - /** Hook called after {@link #execute(String)}. Override for cleanup. */ - protected void destroy() { + super(parameter, tableEnv); } /** @@ -87,43 +62,16 @@ protected void destroy() { * #handle()}, so this returns {@code null} by default — override if a specific job result is * needed. */ - protected JobExecutionResult execute(String jobName) throws Exception { - // TODO: replace with Utils.printLogo(...) once streampark-common Java migration (Phase 1) lands - System.out.println("[StreamPark] FlinkTable " + jobName + " Starting..."); + @Override + protected JobExecutionResult execute(String jobName) { + // TODO(#4408): replace with Utils.printLogo(...) once streampark-common Java migration + // (Phase 1) lands + LOG.info("[StreamPark] FlinkTable {} Starting...", jobName); return null; } - /** - * Convenience shortcut for running a single SQL statement, matching legacy {@code sql(...)}. - * Statement result lines (e.g. from {@code SHOW TABLES}, {@code EXPLAIN}) are logged by - * default — use {@link #sql(String, Consumer)} to receive them instead. - */ - public void sql(String sql) { - sql(sql, null); - } - - /** - * Runs a single SQL statement, routing any result lines to the given callback instead of the - * default log output. - * - *

{@code FlinkSqlExecutor.executeSql} (not yet migrated off Scala — see Phase 3) takes a - * Scala {@code String => Unit} as its 4th argument; this bridges a plain Java {@link Consumer} - * to that type so the public surface of this class stays Scala-free. - */ - public void sql(String sql, Consumer callback) { - FlinkJobSupport.executeSql(sql, parameter, tableEnv, callback); - } - /** Direct access to the underlying Table API — use this instead of delegate methods. */ public TableEnvironment getTableEnv() { return tableEnv; } - - public ParameterTool getParameter() { - return parameter; - } - - private String getAppName() { - return FlinkJobSupport.requireAppName(parameter); - } }