[Feature][Flink] Define FlinkTableJob and FlinkStreamTableJob Java base classes

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.
+ *
+ * <p>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<String> callback) {
+        FlinkJobSupport.executeSql(sql, parameter, tableEnv, callback);
+    }
+
+    public ParameterTool getParameter() {
+        return parameter;
+    }
+}

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.
+     *
+     * <p>{@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<String> callback) {
+        Function1<String, BoxedUnit> scalaCallback =
+            callback == null
+                ? null
+                : new AbstractFunction1<String, BoxedUnit>() {
+
+                    @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.
+     *
+     * <p>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;
+    }
+}

streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkStreamTableJob.java

@@ -0,0 +1,93 @@
+/*
+ * 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.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Java lifecycle base class for jobs that mix the DataStream API and the Table API.
+ *
+ * <p>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).
+ *
+ * <p><b>Note on {@code toDataStream} conversions:</b> 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 extends AbstractFlinkJob {
+
+    private static final Logger LOG = LoggerFactory.getLogger(FlinkStreamTableJob.class);
+
+    protected final StreamExecutionEnvironment env;
+
+    private boolean convertedToDataStream = false;
+
+    protected FlinkStreamTableJob(
+                                  ParameterTool parameter, StreamExecutionEnvironment env,
+                                  StreamTableEnvironment tableEnv) {
+        super(parameter, tableEnv);
+        this.env = env;
+    }
+
+    /**
+     * 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()}.
+     */
+    @Override
+    protected JobExecutionResult execute(String jobName) throws Exception {
+        // 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;
+    }
+
+    /**
+     * 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 (StreamTableEnvironment) tableEnv;
+    }
+}

streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/javaapi/FlinkTableJob.java

@@ -0,0 +1,77 @@
+/*
+ * 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Java lifecycle base class for Flink Table API (batch) jobs.
+ *
+ * <p>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.
+ *
+ * <p>Typical usage:
+ *
+ * <pre>{@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");
+ *   }
+ * }
+ * }</pre>
+ */
+public abstract class FlinkTableJob extends AbstractFlinkJob {
+
+    private static final Logger LOG = LoggerFactory.getLogger(FlinkTableJob.class);
+
+    protected FlinkTableJob(ParameterTool parameter, TableEnvironment tableEnv) {
+        super(parameter, tableEnv);
+    }
+
+    /**
+     * 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.
+     */
+    @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;
+    }
+
+    /** Direct access to the underlying Table API — use this instead of delegate methods. */
+    public TableEnvironment getTableEnv() {
+        return tableEnv;
+    }
+}