{"record":{"id":"bcf2436facd3353e","repo":"apache/seatunnel","slug":"the-transform-sink-s-is-not-configured-with-s","errorCode":null,"errorMessage":"The transform/sink(%s) is not configured with '%s' option","messagePattern":"The transform/sink\\((.+?)\\) 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":247,"sourceCode":"                });\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        }\n    }\n\n    private static String getTableId(ReadonlyConfig config) {\n        return config.getOptional(PLUGIN_OUTPUT).orElse(DEFAULT_ID);\n    }\n\n    static List<String> getInputIds(ReadonlyConfig config) {\n        return config.getOptional(PLUGIN_INPUT).orElse(Collections.singletonList(DEFAULT_ID));\n    }\n\n    public static String getFactoryId(ReadonlyConfig readonlyConfig) {\n        String pluginName = readonlyConfig.get(PLUGIN_NAME);","sourceCodeStart":229,"sourceCodeEnd":265,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/ConfigParserUtil.java#L229-L265","documentation":"checkExistInputTableId requires every transform/sink in a complex graph to declare a plugin_input id. If absent, JobDefineCheckException wrapping an OptionValidationException for PLUGIN_INPUT is thrown. Consumers in a multi-table DAG must state which upstream table they read.","triggerScenarios":"A transform or sink block in a complex multi-table job omits plugin_input (or source_table_name); enforced by checkExistInputTableId inside checkComplexGraph.","commonSituations":"Simple single-table config (where plugin_input is optional) grew into a multi-table graph by adding a second source; plugin blocks copied without adding input ids; ambiguous consumer after renaming tables.","solutions":["Add plugin_input = \"existing-table-id\" to every transform and sink block","Alternatively use the legacy source_table_name option if supported","Ensure the referenced id matches a plugin_output of an upstream plugin","Check the plugin name in the message to locate the offending block"],"exampleFix":"// before\nsink {\n  Console {}\n}\n// after\nsink {\n  Console { plugin_input = \"fake\" }\n}","handlingStrategy":"validation","validationCode":"for (Config c : transformsAndSinks) {\n    if (!c.hasPath(\"plugin_input\") && !c.hasPath(\"source_table_name\")) {\n        throw new IllegalArgumentException(\"Missing plugin_input for \" + c.getString(\"plugin_name\"));\n    }\n}","typeGuard":null,"tryCatchPattern":"try {\n    ConfigParserUtil.checkExistInputTableId(configs);\n} catch (JobDefineCheckException e) {\n    log.error(\"Missing plugin_input: {}\", e.getMessage());\n    throw e;\n}","preventionTips":["Add plugin_input to every transform/sink in multi-table jobs","Use a config template that includes the field","Verify ids match an upstream plugin_output","Lint configs before submission"],"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"}