{"record":{"id":"013471fc6b5b7199","repo":"apache/seatunnel","slug":"commonerrorcode-illegal-argument-013471","errorCode":"CommonErrorCode.ILLEGAL_ARGUMENT","errorMessage":"Topic must be configured when SeaTunnelRow.getTableId() is null","messagePattern":"Topic must be configured when SeaTunnelRow\\.getTableId\\(\\) is null","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/PulsarSinkWriter.java","lineNumber":148,"sourceCode":"                                        pluginConfig,\n                                        messageRoutingMode);\n        this.pendingMessages = new AtomicLong(0);\n        this.sendMessageException = new AtomicReference<>();\n\n        if (PulsarSemantics.EXACTLY_ONCE == pulsarSemantics) {\n            this.transaction = createTransaction();\n        }\n    }\n\n    String resolveTopic(SeaTunnelRow row) {\n        String tableId = row.getTableId();\n        if (tableId != null && !tableId.isEmpty()) {\n            return tableId;\n        }\n\n        String topic = pluginConfig.get(PulsarSinkOptions.TOPIC);\n        if (topic == null || topic.isEmpty()) {\n            throw new PulsarConnectorException(\n                    CommonErrorCode.ILLEGAL_ARGUMENT,\n                    \"Topic must be configured when SeaTunnelRow.getTableId() is null\");\n        }\n\n        return topic;\n    }\n\n    Producer<byte[]> getOrCreateProducer(String topic) {\n        Producer<byte[]> existing = producerMap.get(topic);\n        if (existing != null) {\n            return existing;\n        }\n\n        try {\n            Producer<byte[]> producer = producerCreator.create(topic);\n\n            producerMap.put(topic, producer);\n            return producer;","sourceCodeStart":130,"sourceCodeEnd":166,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java#L130-L166","documentation":"PulsarSinkWriter.resolveTopic determines the target topic per row: first the row's tableId, then the configured 'topic' option. If both the row's tableId is null/empty and no topic option is configured, the writer cannot route the row and throws this IllegalArgumentException-style connector error.","triggerScenarios":"Writing a SeaTunnelRow whose getTableId() is null or empty while PulsarSinkOptions.TOPIC is unset in pluginConfig (resolveTopic, PulsarSinkWriter.java:148).","commonSituations":"Upstream transform/source does not set tableId; multi-table config omits the global topic fallback; tests call the writer without either field.","solutions":["Set the 'topic' option in the sink config so rows without tableId have a fallback.","Ensure the upstream source/transform sets SeaTunnelRow.tableId (e.g. use a TableTransform or catalog table metadata).","If both are intentionally empty for multi-table, add the missing per-table topic mapping."],"exampleFix":"// before\nsink {\n  Pulsar {\n    serviceUrl = \"pulsar://localhost:6650\"\n    // no topic, rows carry no tableId\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 ((row.getTableId() == null || row.getTableId().isEmpty()) && !config.getOptional(PulsarSinkOptions.TOPIC).isPresent()) {\n    throw new IllegalArgumentException(\"Either row.tableId or sink 'topic' must be set\");\n}","typeGuard":"boolean hasTopicOrTableId = (row -> row.getTableId() != null && !row.getTableId().isEmpty()) || hasTopicConfig;","tryCatchPattern":"try {\n    writer.write(row);\n} catch (PulsarConnectorException e) {\n    if (CommonErrorCode.ILLEGAL_ARGUMENT.equals(e.getSeaTunnelErrorCode())) {\n        log.error(\"Row has no tableId and no default topic configured\");\n    }\n    throw e;\n}","preventionTips":["Set a default 'topic' option whenever rows may lack tableId.","Ensure upstream sources/transforms populate SeaTunnelRow.tableId in multi-table pipelines.","Write unit tests covering rows without tableId."],"tags":["pulsar","sink","topic-resolution"],"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"}