[Fix][Zeta] Resolve transform dependencies without hangs or dropped inputs

docs/en/introduction/concepts/incompatible-changes.md

@@ -298,6 +298,31 @@ The Zeta SQL ARRAY function now declares TINYINT elements as `ARRAY<TINYINT>`, m
   - **Migration Guide**: No configuration or state migration is required. Review
     downstream expectations that depended on the previous rounded results.
 
+### Transform Dependency Resolution
+
+- **Behavior change: Reject unresolved inputs in multi-transform jobs**
+  - **Affected components**: Zeta job configuration parsing and `--dry-run connect`.
+  - **Description**: Transforms now wait for all declared `plugin_input` dependencies. A
+    multi-transform configuration with unresolved inputs is rejected instead of silently
+    dropping unavailable inputs or connecting the last unresolved transform to an unintended
+    upstream table. Cyclic or unresolved dependency graphs that previously caused an
+    indefinite retry loop now fail with `JobDefineCheckException` during Zeta parsing or
+    `ConfigCheckException` during connect dry-run validation. An explicit self-reference with
+    no previously available input is rejected too, including single-transform jobs.
+  - **Migration guide**: Correct `plugin_input` references to match the intended source or
+    transform `plugin_output`, and remove dependency cycles. The same validation runs when
+    resubmitting a job from a savepoint, so correct partially resolved or unresolved `plugin_input`
+    before resubmission, apart from the retained legacy fallbacks below. A corrected or valid
+    out-of-order graph now feeds previously dropped configured upstreams to the transform;
+    review the resulting downstream output. Reordering transforms is not
+    required for an otherwise valid dependency graph. The existing single-transform mismatch
+    fallback and terminal explicit-empty-input fallback are retained. An omitted `plugin_input`
+    still resolves the default output ID first; it does not always select the preceding transform.
+    When that ID is unavailable and only the omitted-input transform remains, the parser retains
+    the legacy fallback to the last inserted table. Graph validation's existing requirements for
+    explicit IDs in complex jobs are unchanged. Valid graphs whose dependencies resolved correctly
+    retain their evaluation order and default transform action names. (#12079)
+
 ### Redis Authentication
 
 - Redis sources and sinks now authenticate as the configured nonblank `user` in both `SINGLE` and

docs/zh/introduction/concepts/incompatible-changes.md

@@ -260,6 +260,13 @@ Zeta SQL ARRAY 函数现在将 TINYINT 元素声明为 `ARRAY<TINYINT>`，与实
     元素。声明的输出类型、输入类型转换、null 处理和浮点排序规则保持不变。
   - **迁移指南**：不需要迁移配置或状态。请检查依赖旧舍入结果的下游预期值。
 
+### Transform 依赖解析
+
+- **行为变更：拒绝多 Transform 作业中无法解析的输入**
+  - **影响范围**：Zeta 作业配置解析和 `--dry-run connect`。
+  - **变更说明**：Transform 现在会等待所有声明的 `plugin_input` 依赖就绪。多 Transform 配置中存在无法解析的输入时，会被拒绝，而不再静默丢弃不可用的输入，或将最后一个未解析的 Transform 连接到非预期的上游表。此前导致无限重试的依赖图，现在会在 Zeta 解析时报 `JobDefineCheckException`，或在 connect dry-run 校验时报 `ConfigCheckException`。没有已就绪输入的显式自引用也会被拒绝，包括单 Transform 作业。
+  - **迁移指南**：修正 `plugin_input`，使其引用预期的 Source 或 Transform 的 `plugin_output`，并消除循环依赖。从 savepoint 重新提交作业时也会执行相同校验，因此除下述保留的旧回退逻辑外，重新提交前必须修正部分可解析或无法解析的 `plugin_input`。修正后的依赖图或依赖有效但声明顺序颠倒的图，现在会将此前静默丢弃的已配置上游输入传给 Transform，请检查由此产生的下游输出变化。对于依赖关系本身有效的图，无需调整 Transform 的声明顺序。原有的单 Transform 输入不匹配回退逻辑和末尾显式空输入列表的回退逻辑仍然保留。省略 `plugin_input` 时仍优先解析默认输出 ID，而不是始终选择前一个 Transform。只有默认 ID 不可用且仅剩该省略输入的 Transform 时，解析器才回退到最后插入的表。复杂作业图校验对显式 ID 的原有要求不变。此前能够正确解析依赖的有效图，其求值顺序和默认 Transform Action 名称保持不变。 (#12079)
+
 ### Redis 认证
 
 - Redis Source 和 Sink 现在会在 `SINGLE` 和 `CLUSTER` 模式下以非空白的 `user` 指定的用户认证。

seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/DryRunConnectValidator.java

@@ -37,7 +37,10 @@
 import org.apache.seatunnel.common.constants.PluginType;
 import org.apache.seatunnel.common.utils.DryRunConnectFailureMessageSanitizer;
 import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
+import org.apache.seatunnel.engine.common.exception.JobDefineCheckException;
 import org.apache.seatunnel.engine.core.parse.ConfigParserUtil;
+import org.apache.seatunnel.engine.core.parse.TransformDependencyScheduler;
+import org.apache.seatunnel.engine.core.parse.TransformDependencyScheduler.ScheduledTransform;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -46,16 +49,15 @@
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Queue;
+import java.util.Set;
 import java.util.stream.Collectors;
 
-import static org.apache.seatunnel.api.options.ConnectorCommonOptions.PLUGIN_INPUT;
 import static org.apache.seatunnel.api.options.ConnectorCommonOptions.PLUGIN_OUTPUT;
 import static org.apache.seatunnel.api.table.factory.FactoryUtil.DEFAULT_ID;
+import static org.apache.seatunnel.engine.core.parse.ConfigParserUtil.getInputIds;
 
 /**
  * Performs Layer 1 ({@code --dry-run connect}) validation without creating source/sink runtime
@@ -209,57 +211,65 @@ private void validateTransforms(
         if (configs.isEmpty()) {
             return;
         }
-        Queue<Config> remainingTransforms = new LinkedList<>(configs);
-        int index = 0;
-        while (!remainingTransforms.isEmpty()) {
-            Config transformConfig = remainingTransforms.poll();
-            if (!validateTransform(
-                    index++,
-                    transformConfig,
-                    remainingTransforms,
-                    classLoader,
-                    tableWithSchemas,
-                    results)) {
-                index--;
-                remainingTransforms.offer(transformConfig);
-            }
+        List<ScheduledTransform> scheduledTransforms =
+                scheduleTransforms(configs, tableWithSchemas.keySet());
+        int evaluationIndex = 0;
+        for (ScheduledTransform scheduledTransform : scheduledTransforms) {
+            results.add(
+                    validateTransform(
+                            evaluationIndex++, scheduledTransform, classLoader, tableWithSchemas));
+        }
+    }
+
+    /**
+     * Uses the runtime dependency scheduler and translates job-definition failures into dry-run
+     * configuration errors, preserving the original message and cause.
+     */
+    static List<ScheduledTransform> scheduleTransforms(
+            List<? extends Config> configs, Set<String> initialOutputIds) {
+        try {
+            return TransformDependencyScheduler.scheduleTransforms(configs, initialOutputIds);
+        } catch (JobDefineCheckException e) {
+            throw new ConfigCheckException(e.getMessage(), e);
         }
     }
 
-    private boolean validateTransform(
+    private PluginResult validateTransform(
             int configIndex,
-            Config transformConfig,
-            Queue<Config> remainingTransforms,
+            ScheduledTransform scheduledTransform,
             ClassLoader classLoader,
-            LinkedHashMap<String, SchemaInfo> tableWithSchemas,
-            List<PluginResult> results) {
-        ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(transformConfig);
+            LinkedHashMap<String, SchemaInfo> tableWithSchemas) {
+        ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(scheduledTransform.getConfig());
         String factoryId = ConfigParserUtil.getFactoryId(readonlyConfig);
-        String outputId = readonlyConfig.getOptional(PLUGIN_OUTPUT).orElse(DEFAULT_ID);
+        String outputId = scheduledTransform.getOutputId();
         try {
-            List<SchemaInfo> inputSchemas =
-                    getInputIds(readonlyConfig).stream()
-                            .map(tableWithSchemas::get)
-                            .filter(Objects::nonNull)
-                            .collect(Collectors.toList());
-            if (inputSchemas.isEmpty()) {
-                if (remainingTransforms.isEmpty()) {
-                    inputSchemas = Collections.singletonList(findLast(tableWithSchemas));
-                } else {
-                    // Upstream transform not resolved yet; retry after the rest of the queue.
-                    return false;
+            List<String> inputIds = scheduledTransform.getInputIds();
+            List<SchemaInfo> inputSchemas;
+            if (scheduledTransform.isLegacyFallback()) {
+                inputSchemas = Collections.singletonList(findLast(tableWithSchemas));
+            } else {
+                List<String> missingInputIds =
+                        inputIds.stream()
+                                .filter(inputId -> !tableWithSchemas.containsKey(inputId))
+                                .collect(Collectors.toList());
+                if (!missingInputIds.isEmpty()) {
+                    throw new ConfigCheckException(
+                            "Transform '"
+                                    + outputId
+                                    + "' is missing scheduled inputs "
+                                    + missingInputIds);
                 }
+                inputSchemas =
+                        inputIds.stream().map(tableWithSchemas::get).collect(Collectors.toList());
             }
 
             if (inputSchemas.stream().anyMatch(SchemaInfo::isUnknown)) {
                 tableWithSchemas.put(outputId, SchemaInfo.unknown());
-                results.add(
-                        PluginResult.skipped(
-                                PluginType.TRANSFORM,
-                                configIndex,
-                                factoryId,
-                                "upstream schema not available; schema wiring NOT verified"));
-                return true;
+                return PluginResult.skipped(
+                        PluginType.TRANSFORM,
+                        configIndex,
+                        factoryId,
+                        "upstream schema not available; schema wiring NOT verified");
             }
 
             List<CatalogTable> inputCatalogTables =
@@ -277,13 +287,8 @@ private boolean validateTransform(
                             factoryId);
             tableWithSchemas.put(
                     outputId, SchemaInfo.trusted(transform.getProducedCatalogTables()));
-            results.add(
-                    PluginResult.validated(
-                            PluginType.TRANSFORM,
-                            configIndex,
-                            factoryId,
-                            "schema wiring validated"));
-            return true;
+            return PluginResult.validated(
+                    PluginType.TRANSFORM, configIndex, factoryId, "schema wiring validated");
         } catch (Exception e) {
             throw wrap(PluginType.TRANSFORM, configIndex, factoryId, e);
         }
@@ -438,10 +443,6 @@ private void checkCatalogTableTypesEqual(
         }
     }
 
-    private List<String> getInputIds(ReadonlyConfig config) {
-        return config.getOptional(PLUGIN_INPUT).orElse(Collections.singletonList(DEFAULT_ID));
-    }
-
     private <T> T findLast(LinkedHashMap<?, T> map) {
         if (map.isEmpty()) {
             throw new ConfigCheckException(

seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/DryRunTestSinkFactory.java

@@ -35,6 +35,8 @@ public class DryRunTestSinkFactory
 
     static final Option<Boolean> FAIL_VALIDATION =
             Options.key("fail_validation").booleanType().defaultValue(false);
+    static final Option<String> EXPECTED_TABLE =
+            Options.key("expected_table").stringType().noDefaultValue();
 
     @Override
     public String factoryIdentifier() {
@@ -43,13 +45,27 @@ public String factoryIdentifier() {
 
     @Override
     public OptionRule optionRule() {
-        return OptionRule.builder().optional(FAIL_VALIDATION).build();
+        return OptionRule.builder().optional(FAIL_VALIDATION, EXPECTED_TABLE).build();
     }
 
     @Override
     public void validateConnectionForDryRun(TableSinkFactoryContext context) {
         if (context.getOptions().get(FAIL_VALIDATION)) {
             throw new IllegalStateException("simulated sink validation failure: target missing");
         }
+        context.getOptions()
+                .getOptional(EXPECTED_TABLE)
+                .ifPresent(
+                        expectedTable -> {
+                            String actualTable =
+                                    context.getCatalogTable().getTableId().getTableName();
+                            if (!expectedTable.equals(actualTable)) {
+                                throw new IllegalStateException(
+                                        "expected table "
+                                                + expectedTable
+                                                + " but received "
+                                                + actualTable);
+                            }
+                        });
     }
 }

seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/DryRunTestTransformFactory.java

@@ -0,0 +1,120 @@
+/*
+ * 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.seatunnel.core.starter.seatunnel.command;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.connector.TableTransform;
+import org.apache.seatunnel.api.table.factory.TableTransformFactory;
+import org.apache.seatunnel.api.table.factory.TableTransformFactoryContext;
+import org.apache.seatunnel.api.transform.SeaTunnelTransform;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** Test-only transform factory that verifies how many distinct input schemas reach the factory. */
+public class DryRunTestTransformFactory implements TableTransformFactory {
+
+    static final Option<Integer> EXPECTED_INPUT_COUNT =
+            Options.key("expected_input_count").intType().noDefaultValue();
+    static final Option<String> PRODUCED_TABLE =
+            Options.key("produced_table").stringType().noDefaultValue();
+    static final Option<String> EXPECTED_INPUT_TABLE =
+            Options.key("expected_input_table").stringType().noDefaultValue();
+
+    private static final List<String> CREATED_TABLES = new ArrayList<>();
+
+    @Override
+    public String factoryIdentifier() {
+        return "DryRunTestTransform";
+    }
+
+    @Override
+    public OptionRule optionRule() {
+        return OptionRule.builder()
+                .required(EXPECTED_INPUT_COUNT, PRODUCED_TABLE)
+                .optional(EXPECTED_INPUT_TABLE)
+                .build();
+    }
+
+    @Override
+    public TableTransform<?> createTransform(TableTransformFactoryContext context) {
+        int expectedInputCount = context.getOptions().get(EXPECTED_INPUT_COUNT);
+        if (context.getCatalogTables().size() != expectedInputCount) {
+            throw new IllegalStateException(
+                    "expected "
+                            + expectedInputCount
+                            + " input schemas but received "
+                            + context.getCatalogTables().size());
+        }
+        String producedTableName = context.getOptions().get(PRODUCED_TABLE);
+        context.getOptions()
+                .getOptional(EXPECTED_INPUT_TABLE)
+                .ifPresent(
+                        expected -> {
+                            String actual =
+                                    context.getCatalogTables().get(0).getTableId().getTableName();
+                            if (!expected.equals(actual)) {
+                                throw new IllegalStateException(
+                                        "expected input table "
+                                                + expected
+                                                + " but received "
+                                                + actual);
+                            }
+                        });
+        CatalogTable producedTable =
+                CatalogTableUtil.getCatalogTable(
+                        producedTableName, context.getCatalogTables().get(0).getSeaTunnelRowType());
+        synchronized (CREATED_TABLES) {
+            CREATED_TABLES.add(producedTableName);
+        }
+        return () ->
+                new SeaTunnelTransform<Object>() {
+                    @Override
+                    public CatalogTable getProducedCatalogTable() {
+                        return producedTable;
+                    }
+
+                    @Override
+                    public List<CatalogTable> getProducedCatalogTables() {
+                        return Collections.singletonList(producedTable);
+                    }
+
+                    @Override
+                    public String getPluginName() {
+                        return "DryRunTestTransform";
+                    }
+                };
+    }
+
+    static void resetCreatedTables() {
+        synchronized (CREATED_TABLES) {
+            CREATED_TABLES.clear();
+        }
+    }
+
+    static List<String> getCreatedTables() {
+        synchronized (CREATED_TABLES) {
+            return new ArrayList<>(CREATED_TABLES);
+        }
+    }
+}