alibaba/canal · error · CanalClientException

Receive pulsar batch message error

Error message

Receive pulsar batch message error

What it means

Thrown by CanalPulsarMQConsumer.getMessage when pulsarMQConsumer.batchReceive() raises a PulsarClientException during steady-state consumption. Unlike subscribe-time errors, this happens after the consumer is active, so the usual cause is a runtime broker/network problem or the consumer having been closed/Disconnected underneath the caller.

Source

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

            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);
                    List<CommonMessage> list = MessageUtil.convert(message);
                    messageList.addAll(list);
                } else {
                    CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class);
                    messageList.add(commonMessage);
                }
            }
        } catch (PulsarClientException e) {
            throw new CanalClientException("Receive pulsar batch message error", e);
        }

        return messageList;
    }

    @Override
    public void rollback() {
        try {
            if (isConsumerActive() && hasLastMessages()) {
                // 回滚所有消息
                this.pulsarMQConsumer.negativeAcknowledge(this.lastGetBatchMessage);
            }
        } finally {
            this.lastGetBatchMessage = null;
        }
    }

    @Override

View on GitHub (pinned to 87be50e876)

Solutions

1. Treat this as retriable: catch CanalClientException, wait briefly, and re-call getMessage (the Pulsar client attempts auto-reconnect).
2. Ensure getMessage and disconnect are not called concurrently; guard with the isConsumerActive() check.
3. If it recurs, verify broker health and that pulsarmq.server.url still resolves/reaches.
4. Re-init the consumer (connect()) if isConsumerActive() returns false.

Example fix

// before
public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {
    try {
        Messages<byte[]> messages = pulsarMQConsumer.batchReceive();
        ...
    } catch (PulsarClientException e) {
        throw new CanalClientException("Receive pulsar batch message error", e);
    }
}

// after — retry once on transient failure, re-init if inactive
int attempts = 0;
while (attempts++ < 2) {
    try {
        if (!isConsumerActive()) { connect(); }
        Messages<byte[]> messages = pulsarMQConsumer.batchReceive();
        ...
        return messageList;
    } catch (PulsarClientException e) {
        if (attempts >= 2) throw new CanalClientException("Receive pulsar batch message error", e);
    }
}

if (!isConsumerActive()) {
    connect(); // re-init if the consumer was closed
}

boolean consumerReady() {
    return pulsarMQConsumer != null && pulsarMQConsumer.isConnected();
}

try {
    Messages<byte[]> messages = pulsarMQConsumer.batchReceive();
    ...
} catch (PulsarClientException e) {
    if (!isConsumerActive()) connect();
    throw new CanalClientException("Receive pulsar batch message error", e);
}

When it happens

Trigger: pulsarMQConsumer.batchReceive() at line 234 throws. Causes: consumer was closed (e.g. disconnect() ran concurrently); broker connection lost and not yet recovered; topic/subscription forcibly terminated by the broker; authorization revoked mid-stream.

Common situations: Long-running consumer where the broker restarted or dropped the connection; concurrent disconnect() during getMessage; network partition; broker-side topic unload causing a transient failure before the client auto-reconnects.

Related errors

• Subscript pulsar consumer error
• Pulsar Consumer subscriptName required
• Disconnect pulsar consumer error
• Start RabbitMQ producer error
• error

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