{"record":{"id":"21d8afaaabda8464","repo":"apache/seatunnel","slug":"no-upstream-source-or-transform-is-available-for-s","errorCode":null,"errorMessage":"No upstream source or transform is available for sink.","messagePattern":"No upstream source or transform is available for sink\\.","errorType":"validation","errorClass":"ConfigCheckException","httpStatus":null,"severity":"error","filePath":"seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/DryRunConnectValidator.java","lineNumber":447,"sourceCode":"                throw new ConfigCheckException(\n                        location(pluginType, configIndex, factoryId)\n                                + \" does not support processing inputs with different schemas. \"\n                                + \"Expected table \"\n                                + expected.getTableId()\n                                + \" but found table \"\n                                + catalogTable.getTableId()\n                                + \".\");\n            }\n        }\n    }\n\n    private List<String> getInputIds(ReadonlyConfig config) {\n        return config.getOptional(PLUGIN_INPUT).orElse(Collections.singletonList(DEFAULT_ID));\n    }\n\n    private <T> T findLast(LinkedHashMap<?, T> map) {\n        if (map.isEmpty()) {\n            throw new ConfigCheckException(\n                    \"No upstream source or transform is available for sink.\");\n        }\n        T result = null;\n        for (T value : map.values()) {\n            result = value;\n        }\n        return result;\n    }\n\n    private void logSummary(List<PluginResult> results) {\n        StringBuilder summary = new StringBuilder(\"Dry-run connect validation summary:\");\n        for (PluginResult result : results) {\n            summary.append(System.lineSeparator()).append(\"  \").append(result);\n        }\n        log.info(summary.toString());\n    }\n\n    private ConfigCheckException wrap(","sourceCodeStart":429,"sourceCodeEnd":465,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/DryRunConnectValidator.java#L429-L465","documentation":"findLast returns the last entry of an ordered map of upstream vertices (used to pick the single input for a transform or sink during dry-run validation). If the map is empty — meaning no source or transform produced an output vertex upstream of the plugin — it throws ConfigCheckException('No upstream source or transform is available for sink.'). It catches configs where a sink (or transform chain) has no reachable upstream producer.","triggerScenarios":"A sink block whose referenced plugin_input id matches no source/transform output, or a sink defined with no source at all in the config; the upstream map built during validation is empty when validateSink/validateTransform calls findLast.","commonSituations":"Typos or case mismatches in plugin_input/plugin_output ids; config file missing the source block entirely; commented-out source while the sink remains.","solutions":["Ensure a source block exists and its plugin_output matches the sink's plugin_input exactly.","Remove or fix invalid plugin_input references; sinks without plugin_input default to the DEFAULT_ID producer.","Run with --dry-run CONNECT after fixing to re-validate the DAG."],"exampleFix":"// before\nsink {\n  Console { plugin_input = src_typo }\n}\n// after\nsource {\n  FakeSource { plugin_output = \"src\"; result_table_name = \"src\" }\n}\nsink {\n  Console { plugin_input = \"src\" }\n}","handlingStrategy":"validation","validationCode":"// every sink/transform must reference an existing upstream output id\nconst outputIds = new Set([...sources, ...transforms].flatMap(p => outputsOf(p)));\nfor (const p of [...transforms, ...sinks]) {\n  for (const input of inputsOf(p)) {\n    if (!outputIds.has(input)) throw new Error(\"No upstream producer for input id: \" + input);\n  }\n}","typeGuard":null,"tryCatchPattern":"try { validateConf(conf, DryRun.CONNECT); } catch (ConfigCheckException e) { if (e.getMessage().contains(\"No upstream source or transform is available\")) { fixPluginInputWiring(conf); } else { throw e; } }","preventionTips":["Keep plugin_output/plugin_input ids consistent (exact strings, no typos)","Never leave a sink/transform block without a source","Lint configs so every plugin_input matches some plugin_output"],"tags":["dry-run","dag","missing-upstream","config-validation"],"backgroundTag":"resource-not-found","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}