alibaba/canal · critical · RuntimeException

Start RocketMQ consumer error

Error message

Start RocketMQ consumer error

What it means

Thrown by RocketMQCanalConnector.subscribe when the DefaultMQPushConsumer fails to subscribe to the topic, register the message listener, or call start(). The original MQClientException is wrapped in a RuntimeException. RocketMQ throws MQClientException for broker connectivity, name-server, ACL, or client-configuration problems.

Source

Thrown at client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java:160

            }
            rocketMQConsumer.subscribe(this.topic, "*");
            rocketMQConsumer.registerMessageListener(new MessageListenerOrderly() {

                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageExts, ConsumeOrderlyContext context) {
                    context.setAutoCommit(true);
                    boolean isSuccess = process(messageExts);
                    if (isSuccess) {
                        return ConsumeOrderlyStatus.SUCCESS;
                    } else {
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                }
            });
            rocketMQConsumer.start();
            connected = true;
        } catch (MQClientException ex) {
            throw new RuntimeException("Start RocketMQ consumer error", ex);
        }
    }

    private boolean process(List<MessageExt> messageExts) {
        if (logger.isDebugEnabled()) {
            logger.debug("Get Message: {}", messageExts);
        }
        List messageList = new ArrayList<>();
        for (MessageExt messageExt : messageExts) {
            byte[] data = messageExt.getBody();
            if (data != null) {
                try {
                    if (!flatMessage) {
                        Message message = CanalMessageDeserializer.deserializer(data);
                        messageList.add(message);
                    } else {
                        FlatMessage flatMessage = JSON.parseObject(data, FlatMessage.class);
                        messageList.add(flatMessage);

View on GitHub (pinned to 87be50e876)

Solutions

1. Verify nameServer (rocketMQConsumer.setNamesrvAddr) is reachable: telnet/curl the nameserver port from the client host.
2. Inspect the wrapped MQClientException cause — its error code/message identifies broker-vs-config-vs-ACL failure.
3. Confirm the topic exists and the consumer group has SUB permission on the broker.
4. If using Aliyun ACL, validate accessKey/secretKey and set accessChannel="cloud"; for self-hosted, leave it unset.
5. Ensure only one consumer instance uses the group, or that concurrent instances are expected (cluster mode).

Example fix

// before
connector.subscribe(); // throws RuntimeException on broker/auth failure
// after
try {
    connector.subscribe();
} catch (RuntimeException e) {
    Throwable cause = (e.getCause() != null) ? e.getCause() : e;
    log.error("RocketMQ subscribe failed: {}", cause.getMessage(), e);
    throw e;
}

// Pre-flight: verify nameserver reachability before subscribing.
try (java.net.Socket s = new java.net.Socket()) {
    s.connect(new java.net.InetSocketAddress(nameServerHost, nameServerPort), 3000);
} catch (IOException e) {
    throw new IllegalStateException("RocketMQ nameserver unreachable: " + nameServer, e);
}

try {
    connector.subscribe();
} catch (RuntimeException e) {
    Throwable cause = e.getCause() instanceof MQClientException ? e.getCause() : e;
    logger.error("RocketMQ subscribe failed: {}", cause.getMessage(), cause);
    // implement startup retry/backoff or fail fast per your topology
    throw e;
}

When it happens

Trigger: Calling subscribe() with a nameServer address that is unreachable or misconfigured; providing a topic that does not exist or the group lacks subscription permission; ACL (accessKey/secretKey) mismatch against an ACL-enabled broker; the consumer group name colliding or exceeding RocketMQ client limits.

Common situations: Wrong or unreachable namesrvAddr in a new environment; RocketMQ broker not started or firewalled; accessChannel set to "cloud" against a non-Aliyun broker or vice versa; the consumer group already in use with incompatible settings; namespace misconfiguration.

Related errors

• mq get/ack not support concurrent & async ack
• Failed to fetch the data after: {timeout}
• mq not support this method
• canal.adminUser is empty , pls check https://github.com/alib
• canal.adminPasswd is empty , pls check https://github.com/al

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