apache/seatunnel · error · PulsarConnectorException

CommonErrorCode.ILLEGAL_ARGUMENT

CommonErrorCode.ILLEGAL_ARGUMENT

Error message

Topic must be configured for single-table Pulsar sink.

What it means

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.

Source

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

                .conditional(
                        PulsarSinkOptions.FORMAT,
                        PulsarSinkOptions.TEXT_FORMAT,
                        PulsarSinkOptions.FIELD_DELIMITER)
                .bundled(PulsarSinkOptions.AUTH_PLUGIN_CLASS, PulsarSinkOptions.AUTH_PARAMS)
                .build();
    }

    @Override
    public TableSink createSink(TableSinkFactoryContext context) {
        validateSingleTableTopic(context);
        return () -> new PulsarSink(context.getOptions(), context.getCatalogTable());
    }

    private void validateSingleTableTopic(TableSinkFactoryContext context) {
        ReadonlyConfig options = context.getOptions();
        if (context.getCatalogTable() != null
                && !options.getOptional(PulsarSinkOptions.TOPIC).isPresent()) {
            throw new PulsarConnectorException(
                    CommonErrorCode.ILLEGAL_ARGUMENT,
                    "Topic must be configured for single-table Pulsar sink.");
        }
    }
}

View on GitHub (pinned to cf67b549a7)

Solutions

1. Add 'topic = "persistent://tenant/namespace/topic"' (or plain topic name) to the sink configuration.
2. Verify the option name matches PulsarSinkOptions.TOPIC ('topic') and is not nested in the wrong block.
3. If multi-table sink is intended, configure tableId->topic routing (topics in table configs) instead of relying on a single topic.

Example fix

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

if (catalogTable != null && !options.getOptional(PulsarSinkOptions.TOPIC).isPresent()) {
    throw new IllegalArgumentException("Pulsar sink requires 'topic' for single-table mode");
}

boolean topicConfigured = ReadonlyConfig r -> r.getOptional(PulsarSinkOptions.TOPIC).isPresent();

try {
    sink = factory.createSink(context);
} catch (PulsarConnectorException e) {
    if (CommonErrorCode.ILLEGAL_ARGUMENT.equals(e.getSeaTunnelErrorCode())) {
        log.error("Missing Pulsar sink topic: {}", e.getMessage());
    }
    throw e;
}

When it happens

Trigger: Calling createSink / PulsarSinkFactory with a non-null CatalogTable while PulsarSinkOptions.TOPIC is absent from the config (validateSingleTableTopic at line 78).

Common situations: 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.

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

• CONFIG_VALIDATION_FAILED
• CONFIG_VALIDATION_FAILED
• SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED
• Failed to close Pulsar client after aborting transactions.
• CommonErrorCode.ILLEGAL_ARGUMENT

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