alibaba/canal · error · RuntimeException

Pulsar Consumer subscriptName required

Error message

Pulsar Consumer subscriptName required

What it means

Thrown by CanalPulsarMQConsumer.init when both the PULSARMQ_SUBSCRIPT_NAME property and the groupId are empty/null. The consumer needs a non-empty subscription name to call builder.subscriptionName(...); without it Pulsar would refuse the subscribe, so the connector fails fast during initialization.

Source

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

    @Override
    public void init(Properties properties, String topic, String groupId) {
        this.topic = topic;
        String flatMessageStr = properties.getProperty(CanalConstants.CANAL_MQ_FLAT_MESSAGE);
        if (StringUtils.isNotEmpty(flatMessageStr)) {
            this.flatMessage = Boolean.parseBoolean(flatMessageStr);
        }
        this.serviceUrl = properties.getProperty(PulsarMQConstants.PULSARMQ_SERVER_URL);
        this.roleToken = properties.getProperty(PulsarMQConstants.PULSARMQ_ROLE_TOKEN);
        this.listenerName = properties.getProperty(PulsarMQConstants.PULSARMQ_LISTENER_NAME);
        this.subscriptName = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME);
        // 采用groupId作为subscriptName，避免所有的都是同一个订阅者名称
        if (StringUtils.isEmpty(this.subscriptName)) {
            this.subscriptName = groupId;
        }

        if (StringUtils.isEmpty(this.subscriptName)) {
            throw new RuntimeException("Pulsar Consumer subscriptName required");
        }
        String batchSizeStr = properties.getProperty(CanalConstants.CANAL_MQ_CANAL_BATCH_SIZE);
        if (StringUtils.isNotEmpty(batchSizeStr)) {
            this.batchSize = Integer.parseInt(batchSizeStr);
        }
        String getBatchTimeoutSecondsStr = properties.getProperty(PulsarMQConstants.PULSARMQ_GET_BATCH_TIMEOUT_SECONDS);
        if (StringUtils.isNotEmpty(getBatchTimeoutSecondsStr)) {
            this.getBatchTimeoutSeconds = Integer.parseInt(getBatchTimeoutSecondsStr);
        }
        String batchProcessTimeoutStr = properties.getProperty(PulsarMQConstants.PULSARMQ_BATCH_PROCESS_TIMEOUT);
        if (StringUtils.isNotEmpty(batchProcessTimeoutStr)) {
            this.batchProcessTimeout = Integer.parseInt(batchProcessTimeoutStr);
        }
        String redeliveryDelaySecondsStr = properties.getProperty(PulsarMQConstants.PULSARMQ_REDELIVERY_DELAY_SECONDS);
        if (StringUtils.isNotEmpty(redeliveryDelaySecondsStr)) {
            this.redeliveryDelaySeconds = Integer.parseInt(redeliveryDelaySecondsStr);
        }
        String ackTimeoutSecondsStr = properties.getProperty(PulsarMQConstants.PULSARMQ_ACK_TIMEOUT_SECONDS);

View on GitHub (pinned to 87be50e876)

Solutions

1. Set the Pulsar subscription name explicitly: properties.setProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME, "my-subscription").
2. Or ensure a non-empty groupId is passed to init() — it is used as the subscription name when the property is absent.
3. Validate the property at startup and fail with a clearer message before constructing the consumer.

Example fix

// before
String subscriptName = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME);
if (StringUtils.isEmpty(subscriptName)) subscriptName = groupId; // groupId also empty

// after — require an explicit subscription name with a clear error
if (StringUtils.isEmpty(this.subscriptName)) {
    throw new IllegalArgumentException(
        "Pulsar subscription name required: set " + PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME
        + " or pass a non-empty groupId");
}

String sub = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME);
if (StringUtils.isEmpty(sub)) sub = groupId;
if (StringUtils.isEmpty(sub)) {
    throw new IllegalStateException(
        "Set " + PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME + " or pass a non-empty groupId");
}

When it happens

Trigger: init(properties, topic, groupId) is called with groupId null/empty AND properties does not contain PULSARMQ_SUBSCRIPT_NAME (pulsarmq.subscriptName). The fallback `subscriptName = groupId` is also empty, so the final isEmpty check at line 126 throws.

Common situations: Pulsar consumer started without setting canal.mq.groupId or pulsarmq.subscriptName; groupId computed dynamically and came back blank; copy-paste config that dropped the subscription property when migrating from another MQ connector.

Related errors

• Subscript pulsar consumer error
• ERROR # The kafka kerberos configuration file does not exist
• Receive pulsar batch message error
• Disconnect pulsar consumer error
• Start RabbitMQ producer error

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