apache/seatunnel · error · PulsarConnectorException

PulsarConnectorErrorCode.OPEN_PULSAR_ADMIN_FAILED

PulsarConnectorErrorCode.OPEN_PULSAR_ADMIN_FAILED

Error message

Failed to create pulsar consumer:

What it means

Thrown by PulsarSplitReaderThread.createPulsarConsumer when consumerBuilder.subscribe() raises a PulsarClientException. The wrapper uses OPEN_PULSAR_ADMIN_FAILED, so despite the name it covers consumer subscription failures: connection, authentication, authorization, or nonexistent topic.

Source

Thrown at seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSplitReaderThread.java:142

    public void committingCursor(MessageId offsetsToCommit) throws PulsarClientException {
        if (consumer == null) {
            consumer = createPulsarConsumer(split);
        }
        consumer.acknowledgeCumulative(offsetsToCommit);
    }

    /** Create a specified {@link Consumer} by the given split information. */
    protected Consumer<byte[]> createPulsarConsumer(PulsarPartitionSplit split) {
        ConsumerBuilder<byte[]> consumerBuilder =
                PulsarConfigUtil.createConsumerBuilder(pulsarClient, consumerConfig);

        consumerBuilder.topic(split.getPartition().getFullTopicName());

        // Create the consumer configuration by using common utils.
        try {
            return consumerBuilder.subscribe();
        } catch (PulsarClientException e) {
            throw new PulsarConnectorException(
                    PulsarConnectorErrorCode.OPEN_PULSAR_ADMIN_FAILED,
                    "Failed to create pulsar consumer:",
                    e);
        }
    }

    /**
     * Closes the Pulsar consumer while exposing the connector classloader to Pulsar cleanup code.
     */
    private void closeConsumer() throws IOException {
        if (consumer != null) {
            try {
                PulsarConfigUtil.runWithConnectorClassLoader(consumer::close);
            } catch (Exception e) {
                throw new IOException("Failed to close Pulsar consumer.", e);
            }
        }
    }

View on GitHub (pinned to cf67b549a7)

Solutions

1. Read the wrapped PulsarClientException cause for the exact failure (connection, authorization, topic-not-found, etc.).
2. Verify serviceUrl, authPlugin/authParams, and TLS settings in the connector configuration.
3. Confirm the topic exists (or enable broker auto-creation) and that the principal has consume permission on it.
4. Check network connectivity from worker nodes to the Pulsar broker (pulsar://6650) and broker health.
5. For subscription conflicts, ensure subscription names are unique per job or switch to Shared/Key_Shared subscription type.

Example fix

// before (example fix: topic auto-created expectation)
source {
  Pulsar {
    serviceUrl = "pulsar://localhost:6650"
    topic = "nonexistent/topic"
  }
}
// after
source {
  Pulsar {
    serviceUrl = "pulsar://broker:6650"
    topic = "public/default/my-topic"  // must exist, or enable broker autoCreate
    auth.params = "{"token":"<valid-jwt>"}"
  }
}

// preflight check before submitting the job
PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(adminUrl).build();
if (!admin.topics().getPartitionedTopicList(namespace).contains(topic)) {
    throw new IllegalArgumentException("Topic does not exist: " + topic);
}

try { reader.open(); } catch (PulsarConnectorException e) { LOG.error("Subscribe failed: {}", e.getCause()); if (e.getCause() instanceof PulsarClientException.ConnectionException) { /* retry with backoff */ } }

When it happens

Trigger: open() or committingCursor() triggers consumer creation and the Pulsar client fails to subscribe: broker unreachable, TLS/auth misconfiguration, topic does not exist (and auto-create disabled), or the subscription name is invalid/already in incompatible use.

Common situations: Wrong serviceUrl in Pulsar config; broker down or firewall blocking 6650; JWT/TLS credentials expired or misconfigured; topic deleted while job runs; exclusive subscription already held by another consumer.

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

• CommonErrorCodeDeprecated.READER_OPERATION_FAILED
• SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED
• PulsarConnectorErrorCode.ACK_CUMULATE_FAILED
• Failed to close Pulsar consumer.
• CREATE_ACTIVEMQ_CLIENT_FAILED

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