alibaba/canal · error · CanalClientException

Subscript pulsar consumer error

Error message

Subscript pulsar consumer error

What it means

Thrown by CanalPulsarMQConsumer.connect when ConsumerBuilder.subscribe() raises a PulsarClientException. By this point the PulsarClient is already built; the failure is specifically about creating the consumer on a topic/subscription — invalid service URL, authentication failure, non-existent topic (when not auto-creating), subscription config conflict, or broker unreachable.

Source

Thrown at connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java:225

                dlqBuilder.deadLetterTopic(dlqTopic);
            }

            // 默认关闭，如果需要重试则开启
            builder.enableRetry(true).deadLetterPolicy(dlqBuilder.build());
        }

        // ack超时
        builder.ackTimeout(this.ackTimeoutSeconds, TimeUnit.SECONDS);

        // pulsar批量获取消息设置
        builder.batchReceivePolicy(new BatchReceivePolicy.Builder().maxNumMessages(this.batchSize)
            .timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS)
            .build());

        try {
            this.pulsarMQConsumer = builder.subscribe();
        } catch (PulsarClientException e) {
            throw new CanalClientException("Subscript pulsar consumer error", e);
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {
        List<CommonMessage> messageList = Lists.newArrayList();
        try {
            Messages<byte[]> messages = pulsarMQConsumer.batchReceive();
            if (null == messages || messages.size() == 0) {
                return messageList;
            }
            // 保存当前消费记录，用于ack和rollback
            this.lastGetBatchMessage = messages;
            for (org.apache.pulsar.client.api.Message<byte[]> msg : messages) {
                byte[] data = msg.getData();
                if (!this.flatMessage) {
                    Message message = CanalMessageSerializerUtil.deserializer(data);

View on GitHub (pinned to 87be50e876)

Solutions

1. Verify pulsarmq.server.url is reachable and is a valid puls:// or http(s) Pulsar service URL.
2. Confirm the topic exists (or broker permits auto-creation) and the topic/tenant/namespace string is correct.
3. Check the role token is valid and not expired when PULSARMQ_ROLE_TOKEN is set.
4. For Failover subscriptions, ensure no other consumer currently holds the subscription, or switch to Shared if concurrent consumers are intended.
5. Retry with backoff — a transient broker outage during startup often clears on reconnect.

Example fix

// before
try {
    this.pulsarMQConsumer = builder.subscribe();
} catch (PulsarClientException e) {
    throw new CanalClientException("Subscript pulsar consumer error", e);
}

// after — preserve the broker error detail for diagnosis
try {
    this.pulsarMQConsumer = builder.subscribe();
} catch (PulsarClientException e) {
    throw new CanalClientException(
        "Subscript pulsar consumer error [url=" + serviceUrl + ", topic=" + topic
        + ", sub=" + subscriptName + "]: " + e.getMessage(), e);
}

// Pre-flight checks before connect()
if (StringUtils.isEmpty(serviceUrl)) throw new IllegalStateException("pulsar serviceUrl missing");
if (StringUtils.isEmpty(topic)) throw new IllegalStateException("pulsar topic missing");
if (StringUtils.isEmpty(subscriptName)) throw new IllegalStateException("pulsar subscriptName missing");

int attempts = 0;
while (attempts++ < 3) {
    try {
        this.pulsarMQConsumer = builder.subscribe();
        break;
    } catch (PulsarClientException e) {
        if (attempts >= 3)
            throw new CanalClientException("Subscript pulsar consumer error", e);
    }
}

When it happens

Trigger: builder.subscribe() at line 223 fails. Causes: wrong/malformed serviceUrl; token auth rejected; topic does not exist and broker auto-creation is disabled; subscription name conflicts with an incompatible existing subscription type; broker/network down after client connect.

Common situations: Misconfigured pulsarmq.server.url; expired or wrong role token; topic name typo or tenant/namespace that does not exist; Failover subscription already held by another active consumer; broker temporarily unavailable during consumer startup.

Related errors

• Pulsar Consumer subscriptName required
• Receive pulsar batch message error
• Start RabbitMQ producer error
• Start RabbitMQ producer error
• Start RocketMQ producer error

AI-assisted analysis of alibaba/canal@87be50e876 (2026-08-14). Data as JSON: /api/errors/e78b69d5ec637e70. Report an issue: GitHub.