alibaba/canal · error · CanalClientException

error

Error message

error

What it means

Thrown by CanalRabbitMQConsumer.connect when channel.basicConsume(queueName, false, consumer) raises an IOException. newConnection/createChannel succeeded, but registering the consumer on the queue failed — most often because the queue does not exist or the channel was closed between creation and the consume call.

Source

Thrown at connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/consumer/CanalRabbitMQConsumer.java:128

        if (connect == null) {
            this.connect();
        }

        Consumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {

                if (body != null) {
                    channel.basicAck(envelope.getDeliveryTag(), process(body));
                }
            }
        };
        try {
            channel.basicConsume(queueName, false, consumer);
        } catch (IOException e) {
            throw new CanalClientException("error", e);
        }
    }

    private boolean process(byte[] messageData) {
        if (logger.isDebugEnabled()) {
            logger.debug("Get Message: {}", new String(messageData));
        }
        List<CommonMessage> messageList = new ArrayList<>();
        if (!flatMessage) {
            Message message = CanalMessageSerializerUtil.deserializer(messageData);
            messageList.addAll(MessageUtil.convert(message));
        } else {
            CommonMessage commonMessage = JSON.parseObject(messageData, CommonMessage.class);
            messageList.add(commonMessage);
        }
        ConsumerBatchMessage<CommonMessage> batchMessage = new ConsumerBatchMessage<>(messageList);
        try {
            messageBlockingQueue.put(batchMessage);

View on GitHub (pinned to 87be50e876)

Solutions

1. Ensure the queue exists on the broker, or declare it (channel.queueDeclare) before basicConsume if the consumer is responsible for creation.
2. Verify the rabbitmq.queue property matches an existing, consumable queue and the user has read permission.
3. If the channel may have errored earlier, recreate the channel before consuming.

Example fix

// before
try {
    channel.basicConsume(queueName, false, consumer);
} catch (IOException e) {
    throw new CanalClientException("error", e);
}

// after — declare-if-missing and a descriptive error
try {
    channel.queueDeclarePassive(queueName); // fails fast if queue missing
    channel.basicConsume(queueName, false, consumer);
} catch (IOException e) {
    throw new CanalClientException("basicConsume failed for queue " + queueName, e);
}

// Ensure the queue exists before consuming
channel.queueDeclarePassive(queueName);

try {
    channel.queueDeclarePassive(queueName);
    channel.basicConsume(queueName, false, consumer);
} catch (IOException e) {
    throw new CanalClientException("basicConsume failed for queue " + queueName, e);
}

When it happens

Trigger: channel.basicConsume() at line 132 throws IOException. Causes: the target queue does not exist (consumer does not declare it); the channel was torn down (channel-level exception from a prior operation); access refused on the queue.

Common situations: Queue name typo (rabbitmq.queue); queue not pre-declared on the broker and the consumer does not declare it; permissions/TopicAuth blocking the consume; a previous channel error left the channel in a closed state.

Related errors

• mq get/ack not support concurrent & async ack
• Failed to fetch the data after: ${timeout}
• failed to parse host
• ERROR # The kafka kerberos configuration file does not exist
• Pulsar Consumer subscriptName required

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