apache/seatunnel · error · PulsarConnectorException

CommonErrorCode.ILLEGAL_ARGUMENT

CommonErrorCode.ILLEGAL_ARGUMENT

Error message

Topic must be configured when SeaTunnelRow.getTableId() is null

What it means

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.

Source

Thrown at seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java:148

                                        pluginConfig,
                                        messageRoutingMode);
        this.pendingMessages = new AtomicLong(0);
        this.sendMessageException = new AtomicReference<>();

        if (PulsarSemantics.EXACTLY_ONCE == pulsarSemantics) {
            this.transaction = createTransaction();
        }
    }

    String resolveTopic(SeaTunnelRow row) {
        String tableId = row.getTableId();
        if (tableId != null && !tableId.isEmpty()) {
            return tableId;
        }

        String topic = pluginConfig.get(PulsarSinkOptions.TOPIC);
        if (topic == null || topic.isEmpty()) {
            throw new PulsarConnectorException(
                    CommonErrorCode.ILLEGAL_ARGUMENT,
                    "Topic must be configured when SeaTunnelRow.getTableId() is null");
        }

        return topic;
    }

    Producer<byte[]> getOrCreateProducer(String topic) {
        Producer<byte[]> existing = producerMap.get(topic);
        if (existing != null) {
            return existing;
        }

        try {
            Producer<byte[]> producer = producerCreator.create(topic);

            producerMap.put(topic, producer);
            return producer;

View on GitHub (pinned to cf67b549a7)

Solutions

1. Set the 'topic' option in the sink config so rows without tableId have a fallback.
2. Ensure the upstream source/transform sets SeaTunnelRow.tableId (e.g. use a TableTransform or catalog table metadata).
3. If both are intentionally empty for multi-table, add the missing per-table topic mapping.

Example fix

// before
sink {
  Pulsar {
    serviceUrl = "pulsar://localhost:6650"
    // no topic, rows carry no tableId
  }
}
// after
sink {
  Pulsar {
    serviceUrl = "pulsar://localhost:6650"
    topic = "persistent://public/default/my-topic"
  }
}

if ((row.getTableId() == null || row.getTableId().isEmpty()) && !config.getOptional(PulsarSinkOptions.TOPIC).isPresent()) {
    throw new IllegalArgumentException("Either row.tableId or sink 'topic' must be set");
}

boolean hasTopicOrTableId = (row -> row.getTableId() != null && !row.getTableId().isEmpty()) || hasTopicConfig;

try {
    writer.write(row);
} catch (PulsarConnectorException e) {
    if (CommonErrorCode.ILLEGAL_ARGUMENT.equals(e.getSeaTunnelErrorCode())) {
        log.error("Row has no tableId and no default topic configured");
    }
    throw e;
}

When it happens

Trigger: Writing a SeaTunnelRow whose getTableId() is null or empty while PulsarSinkOptions.TOPIC is unset in pluginConfig (resolveTopic, PulsarSinkWriter.java:148).

Common situations: Upstream transform/source does not set tableId; multi-table config omits the global topic fallback; tests call the writer without either field.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• Failed to close Pulsar client after aborting transactions.
• CommonErrorCode.ILLEGAL_ARGUMENT
• CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT
• Failed to close Pulsar sink writer.
• MultiTableWriterRunnable can't find writer for tableId:

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/013471fc6b5b7199. Report an issue: GitHub.