{"record":{"id":"50a3df770d8c7335","repo":"apache/seatunnel","slug":"unable-to-create-a-sink-for-identifier-factoryi","errorCode":null,"errorMessage":"Unable to create a sink for identifier '${factoryIdentifier}'.","messagePattern":"Unable to create a sink for identifier '(.+?)'\\.","errorType":"exception","errorClass":"FactoryException","httpStatus":null,"severity":"error","filePath":"seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/FactoryUtil.java","lineNumber":262,"sourceCode":"            }\n\n            TableSinkFactoryContext context =\n                    TableSinkFactoryContext.replacePlaceholderAndCreate(\n                            catalogTable,\n                            config,\n                            classLoader,\n                            tableSinkFactory.excludeTablePlaceholderReplaceKeys());\n            ConfigValidator.of(context.getOptions()).validate(tableSinkFactory.optionRule());\n\n            LOG.info(\n                    \"Create sink '{}' with upstream input catalog-table[database: {}, schema: {}, table: {}]\",\n                    factoryIdentifier,\n                    catalogTable.getTablePath().getDatabaseName(),\n                    catalogTable.getTablePath().getSchemaName(),\n                    catalogTable.getTablePath().getTableName());\n            return tableSinkFactory.createSink(context).createSink();\n        } catch (Throwable t) {\n            throw new FactoryException(\n                    String.format(\n                            \"Unable to create a sink for identifier '%s'.\", factoryIdentifier),\n                    t);\n        }\n    }\n\n    public static <IN, StateT, CommitInfoT, AggregatedCommitInfoT>\n            SeaTunnelSink<IN, StateT, CommitInfoT, AggregatedCommitInfoT> createMultiTableSink(\n                    Map<TablePath, SeaTunnelSink> sinks,\n                    ReadonlyConfig options,\n                    ClassLoader classLoader) {\n        try {\n            TableSinkFactory<IN, StateT, CommitInfoT, AggregatedCommitInfoT> factory =\n                    new MultiTableSinkFactory();\n            MultiTableFactoryContext context =\n                    new MultiTableFactoryContext(options, classLoader, sinks);\n            ConfigValidator.of(context.getOptions()).validate(factory.optionRule());\n            return factory.createSink(context).createSink();","sourceCodeStart":244,"sourceCodeEnd":280,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/FactoryUtil.java#L244-L280","documentation":"FactoryUtil.createAndPrepareSink wraps any Throwable raised while discovering, instantiating, or preparing a Sink in a FactoryException with this message. The underlying cause attached to the exception carries the actual failure reason.","triggerScenarios":"Calling FactoryUtil.createAndPrepareSink with a sink identifier whose factory is absent from the classpath, whose createSink()/prepare() throws, whose option validation fails, or whose table path conversion fails.","commonSituations":"Misspelled sink plugin name in the config; sink connector jar not deployed on all nodes; sink prepare() fails on bad connection settings or invalid table schema mapping.","solutions":["Inspect the exception cause to find the real error inside sink creation.","Confirm the sink identifier (e.g. 'Jdbc', 'Console') matches an installed plugin exactly.","Deploy the sink connector jar to every node's connectors directory.","Check sink option names/required options against the connector docs."],"exampleFix":"// before\nsink {\n  Jdbc-Sink {\n    url = \"jdbc:mysql://localhost/test\"\n  }\n}\n// after\nsink {\n  Jdbc {\n    url = \"jdbc:mysql://localhost/test\"\n    driver = \"com.mysql.cj.jdbc.Driver\"\n  }\n}","handlingStrategy":"try-catch","validationCode":"Optional<TableSinkFactory> f = FactoryUtil.discoverOptionalSinkFactory(classLoader, pluginId);\nif (!f.isPresent()) {\n    throw new IllegalArgumentException(\"Sink plugin not installed: \" + pluginId);\n}","typeGuard":null,"tryCatchPattern":"try {\n    sink = FactoryUtil.createAndPrepareSink(catalogTable, cfg, classLoader, pluginId);\n} catch (FactoryException e) {\n    LOG.error(\"Sink '{}' creation failed: {}\", pluginId, e.getCause());\n    throw e;\n}","preventionTips":["Pre-validate sink options against the factory's optionRule.","Deploy sink connector jars to every node before submitting jobs.","Match identifiers exactly as documented (case-insensitive but spelling matters)."],"tags":["factory","spi","sink-plugin","wrapper"],"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"}