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.
+ *
+ * <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;
+    }
+}
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.
+     *
+     * <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;
+    }
+}
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..60f5677ab4
--- /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,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;
+    }
+}
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..d87fe2f3be
--- /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,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;
+    }
+}
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..c555b27872
--- /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,78 @@
+/*
+ * 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.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 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 {
+
+    @Test
+    void handleGetsDirectAccessToEnvAndTableEnv() throws Exception {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        StreamTableEnvironment tableEnv =
+            StreamTableEnvironment.create(env, EnvironmentSettings.newInstance().inStreamingMode().build());
+
+        FlinkStreamTableJob job =
+            new FlinkStreamTableJob(JobTestParams.withAppName("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(JobTestParams.withAppName("test-stream-table-job"), env, tableEnv) {
+
+                @Override
+                protected void handle() {
+                    Table table = getTableEnv().fromValues("a", "b", "c");
+                    DataStream<Row> stream = getTableEnv().toDataStream(table);
+                    stream.addSink(new SinkFunction<Row>() {
+                    }); // 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..4bc4363cf6
--- /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,117 @@
+/*
+ * 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 TableEnvironment newBatchTableEnv() {
+        return TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
+    }
+
+    @Test
+    void lifecycleRunsInOrderAndExposesTableEnv() throws Exception {
+        List<String> calls = new ArrayList<>();
+        TableEnvironment tableEnv = newBatchTableEnv();
+
+        FlinkTableJob job =
+            new FlinkTableJob(JobTestParams.withAppName("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<String, String> 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<String> 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<String, String> empty = new HashMap<>();
+        FlinkTableJob job =
+            new FlinkTableJob(ParameterTool.fromMap(empty), tableEnv) {
+
+                @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.
+                }
+            };
+
+        assertThrows(IllegalArgumentException.class, job::start);
+    }
+}
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<String, String> map = new HashMap<>();
+        map.put("app.name", appName);
+        return ParameterTool.fromMap(map);
+    }
+}