{"record":{"id":"ed005fbdae6f2e88","repo":"apache/seatunnel","slug":"commonerrorcode-illegal-argument","errorCode":"CommonErrorCode.ILLEGAL_ARGUMENT","errorMessage":"Topic must be configured for single-table Pulsar sink.","messagePattern":"Topic must be configured for single-table Pulsar sink\\.","errorType":"error_code","errorClass":"PulsarConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactory.java","lineNumber":78,"sourceCode":"                .conditional(\n                        PulsarSinkOptions.FORMAT,\n                        PulsarSinkOptions.TEXT_FORMAT,\n                        PulsarSinkOptions.FIELD_DELIMITER)\n                .bundled(PulsarSinkOptions.AUTH_PLUGIN_CLASS, PulsarSinkOptions.AUTH_PARAMS)\n                .build();\n    }\n\n    @Override\n    public TableSink createSink(TableSinkFactoryContext context) {\n        validateSingleTableTopic(context);\n        return () -> new PulsarSink(context.getOptions(), context.getCatalogTable());\n    }\n\n    private void validateSingleTableTopic(TableSinkFactoryContext context) {\n        ReadonlyConfig options = context.getOptions();\n        if (context.getCatalogTable() != null\n                && !options.getOptional(PulsarSinkOptions.TOPIC).isPresent()) {\n            throw new PulsarConnectorException(\n                    CommonErrorCode.ILLEGAL_ARGUMENT,\n                    \"Topic must be configured for single-table Pulsar sink.\");\n        }\n    }\n}\n","sourceCodeStart":60,"sourceCodeEnd":84,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactory.java#L60-L84","documentation":"This error is thrown by the Pulsar sink factory during sink creation validation. When a sink is configured for a single table (CatalogTable present), the 'topic' option must be explicitly set, because the writer has no per-row tableId fallback. Without it, the connector cannot determine which Pulsar topic to publish to.","triggerScenarios":"Calling createSink / PulsarSinkFactory with a non-null CatalogTable while PulsarSinkOptions.TOPIC is absent from the config (validateSingleTableTopic at line 78).","commonSituations":"User writes a single-table Pulsar sink config and forgets 'topic = \"...\"'; config key typo'd or placed under the wrong prefix; config generated programmatically omitting TOPIC.","solutions":["Add 'topic = \"persistent://tenant/namespace/topic\"' (or plain topic name) to the sink configuration.","Verify the option name matches PulsarSinkOptions.TOPIC ('topic') and is not nested in the wrong block.","If multi-table sink is intended, configure tableId->topic routing (topics in table configs) instead of relying on a single topic."],"exampleFix":"// before\nsink {\n  Pulsar {\n    serviceUrl = \"pulsar://localhost:6650\"\n  }\n}\n// after\nsink {\n  Pulsar {\n    serviceUrl = \"pulsar://localhost:6650\"\n    topic = \"persistent://public/default/my-topic\"\n  }\n}","handlingStrategy":"validation","validationCode":"if (catalogTable != null && !options.getOptional(PulsarSinkOptions.TOPIC).isPresent()) {\n    throw new IllegalArgumentException(\"Pulsar sink requires 'topic' for single-table mode\");\n}","typeGuard":"boolean topicConfigured = ReadonlyConfig r -> r.getOptional(PulsarSinkOptions.TOPIC).isPresent();","tryCatchPattern":"try {\n    sink = factory.createSink(context);\n} catch (PulsarConnectorException e) {\n    if (CommonErrorCode.ILLEGAL_ARGUMENT.equals(e.getSeaTunnelErrorCode())) {\n        log.error(\"Missing Pulsar sink topic: {}\", e.getMessage());\n    }\n    throw e;\n}","preventionTips":["Always include 'topic' in single-table Pulsar sink configs.","Use config templates that list required sink options.","Lint job configs for required Pulsar sink options before submission."],"tags":["pulsar","sink","config-validation"],"backgroundTag":"missing-required-config-field","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}