{"record":{"id":"98b132df6e7329f4","repo":"apache/seatunnel","slug":"the-s-option-configured-in-s-is-incorrect-a","errorCode":null,"errorMessage":"The '%s' option configured in [%s] is incorrect, and the source/transform[%s] is not found.","messagePattern":"The '(.+?)' option configured in \\[(.+?)\\] is incorrect, and the source/transform\\[(.+?)\\] is not found\\.","errorType":"validation","errorClass":"JobDefineCheckException","httpStatus":null,"severity":"error","filePath":"seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/ConfigParserUtil.java","lineNumber":206,"sourceCode":"                                            id));\n                        }\n                        return new Tuple2<>(config, VertexStatus.CREATED);\n                    });\n        }\n    }\n\n    private static void checkInputId(\n            List<? extends Config> configs,\n            Map<String, Tuple2<Config, VertexStatus>> vertexStatusMap) {\n        for (Config config : configs) {\n            List<String> inputIds = getInputIds(ReadonlyConfig.fromConfig(config));\n            inputIds.forEach(\n                    inputId ->\n                            vertexStatusMap.compute(\n                                    inputId,\n                                    (id, old) -> {\n                                        if (old == null) {\n                                            throw new JobDefineCheckException(\n                                                    String.format(\n                                                            \"The '%s' option configured in [%s] is incorrect, and the source/transform[%s] is not found.\",\n                                                            PLUGIN_INPUT.key(),\n                                                            config.getString(PLUGIN_NAME.key()),\n                                                            id));\n                                        }\n                                        return new Tuple2<>(old._1(), VertexStatus.LINKED);\n                                    }));\n        }\n    }\n\n    private static void checkLinked(Map<String, Tuple2<Config, VertexStatus>> vertexStatusMap) {\n        vertexStatusMap.forEach(\n                (id, vertex) -> {\n                    if (vertex._2() == VertexStatus.CREATED) {\n                        throw new JobDefineCheckException(\n                                String.format(\n                                        \"The '%s' option configured is incorrect, this table(%s) belonging to source/transform(%s) is not used.\",","sourceCodeStart":188,"sourceCodeEnd":224,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/ConfigParserUtil.java#L188-L224","documentation":"checkInputId verifies that every plugin_input id referenced by a transform/sink actually matches a registered source/transform output id in the DAG. If a referenced id was never produced, JobDefineCheckException is thrown saying the source/transform was not found. This catches dangling edges in multi-table job configs.","triggerScenarios":"A transform or sink config references plugin_input (or source_table_name) with a value that no upstream plugin declares as plugin_output; complex-graph validation via checkComplexGraph.","commonSituations":"Typo in plugin_input value; upstream plugin renamed or removed; plugin_output missing on the intended producer (so its default id differs); multi-table job where the producer block is commented out.","solutions":["Fix the plugin_input value to exactly match an existing plugin_output id","Ensure the upstream plugin sets plugin_output with that exact id","Check for typos/case differences between plugin_output and plugin_input values","Confirm the upstream source/transform block is present and not commented out"],"exampleFix":"// before\nsource { FakeSource { plugin_output = \"src\" } }\nsink { Console { plugin_input = \"sorce\" } }\n// after\nsource { FakeSource { plugin_output = \"src\" } }\nsink { Console { plugin_input = \"src\" } }","handlingStrategy":"validation","validationCode":"Set<String> produced = collectPluginOutputs(config);\nfor (String input : collectPluginInputs(config)) {\n    if (!produced.contains(input)) {\n        throw new IllegalArgumentException(\"Unknown plugin_input: \" + input);\n    }\n}","typeGuard":null,"tryCatchPattern":"try {\n    ConfigParserUtil.checkInputId(...);\n} catch (JobDefineCheckException e) {\n    log.error(\"Dangling plugin_input reference: {}\", e.getMessage());\n    throw e;\n}","preventionTips":["Keep plugin_input values identical to upstream plugin_output","Check for typos and case mismatches","Never delete a producer without updating consumers","Use short consistent table ids"],"tags":["java","job-config","dag","reference-not-found"],"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-14T21:17:11.552Z"}