{"record":{"id":"ec4558f34b8bc6d1","repo":"apache/seatunnel","slug":"the-source-transform-s-is-not-configured-with","errorCode":null,"errorMessage":"The source/transform(%s) is not configured with '%s' option","messagePattern":"The source/transform\\((.+?)\\) is not configured with '(.+?)' option","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":235,"sourceCode":"\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.\",\n                                        PLUGIN_INPUT.key(),\n                                        id,\n                                        vertex._1().getString(PLUGIN_NAME.key())));\n                    }\n                });\n    }\n\n    private static void checkExistTableId(List<? extends Config> configs) {\n        for (Config config : configs) {\n            if (!ReadonlyConfig.fromConfig(config).getOptional(PLUGIN_OUTPUT).isPresent()) {\n                throw new JobDefineCheckException(\n                        String.format(\n                                \"The source/transform(%s) is not configured with '%s' option\",\n                                config.getString(PLUGIN_NAME.key()), PLUGIN_OUTPUT.key()),\n                        new OptionValidationException(PLUGIN_OUTPUT));\n            }\n        }\n    }\n\n    private static void checkExistInputTableId(List<? extends Config> configs) {\n        for (Config config : configs) {\n            if (!ReadonlyConfig.fromConfig(config).getOptional(PLUGIN_INPUT).isPresent()) {\n                throw new JobDefineCheckException(\n                        String.format(\n                                \"The transform/sink(%s) is not configured with '%s' option\",\n                                config.getString(PLUGIN_NAME.key()), PLUGIN_INPUT.key()),\n                        new OptionValidationException(PLUGIN_INPUT));\n            }\n        }","sourceCodeStart":217,"sourceCodeEnd":253,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/ConfigParserUtil.java#L217-L253","documentation":"checkExistTableId requires every source/transform in a complex graph to declare a plugin_output table id. If the option is absent, JobDefineCheckException wrapping an OptionValidationException for PLUGIN_OUTPUT is thrown. In multi-table DAGs, producers must be identifiable by their output table id.","triggerScenarios":"A source or transform block in a complex multi-table job omits plugin_output (or result_table_name); enforced by checkExistTableId inside checkComplexGraph.","commonSituations":"Single-table style config (no result_table_name) extended with a second source, making it a complex graph; older configs predating the option; plugin blocks copied without adding output ids.","solutions":["Add plugin_output = \"unique-table-id\" to every source and transform block","Alternatively use the legacy result_table_name option if supported by the plugin","Verify which plugin is missing the option from the plugin name in the message","Keep ids unique across all producers in the job"],"exampleFix":"// before\nsource {\n  FakeSource {}\n  Jdbc { plugin_output = \"db\" }\n}\n// after\nsource {\n  FakeSource { plugin_output = \"fake\" }\n  Jdbc { plugin_output = \"db\" }\n}","handlingStrategy":"validation","validationCode":"for (Config c : sourcesAndTransforms) {\n    if (!c.hasPath(\"plugin_output\") && !c.hasPath(\"result_table_name\")) {\n        throw new IllegalArgumentException(\"Missing plugin_output for \" + c.getString(\"plugin_name\"));\n    }\n}","typeGuard":null,"tryCatchPattern":"try {\n    ConfigParserUtil.checkExistTableId(configs);\n} catch (JobDefineCheckException e) {\n    log.error(\"Missing plugin_output: {}\", e.getMessage());\n    throw e;\n}","preventionTips":["Add plugin_output to every source/transform in multi-table jobs","Use a config template that includes the field","Document the requirement for complex DAGs","Lint configs for the option before submit"],"tags":["java","job-config","dag","missing-option"],"backgroundTag":"missing-required-option","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"}