alibaba/canal · error · CanalClientException

mq get/ack not support concurrent & async ack

Error message

mq get/ack not support concurrent & async ack

What it means

Thrown by CanalRocketMQConsumer.getMessage when a previous batch (lastGetBatchMessage) has not been acknowledged or rolled back before the next getMessage call. The RocketMQ connector mirrors the RabbitMQ one: it enforces strict sequential get→ack/rollback ordering and does not support concurrent or asynchronous acknowledgement.

Source

Thrown at connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/consumer/CanalRocketMQConsumer.java:187

            logger.error("Put message to queue error", e);
            throw new RuntimeException(e);
        }
        boolean isCompleted;
        try {
            isCompleted = batchMessage.waitFinish(batchProcessTimeout);
        } catch (InterruptedException e) {
            logger.error("Interrupted when waiting messages to be finished.", e);
            throw new RuntimeException(e);
        }
        boolean isSuccess = batchMessage.isSuccess();
        return isCompleted && isSuccess;
    }

    @Override
    public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {
        try {
            if (this.lastGetBatchMessage != null) {
                throw new CanalClientException("mq get/ack not support concurrent & async ack");
            }

            ConsumerBatchMessage<CommonMessage> batchMessage = messageBlockingQueue.poll(timeout, unit);
            if (batchMessage != null) {
                this.lastGetBatchMessage = batchMessage;
                return batchMessage.getData();
            }
        } catch (InterruptedException ex) {
            logger.warn("Get message timeout", ex);
            throw new CanalClientException("Failed to fetch the data after: " + timeout);
        }
        return null;
    }

    @Override
    public void rollback() {
        try {
            if (this.lastGetBatchMessage != null) {

View on GitHub (pinned to 87be50e876)

Solutions

1. Always call ack() or rollback() before the next getMessage().
2. Run getMessage/ack/rollback on a single thread or under a lock — the connector is not concurrency-safe.
3. Use try/finally to rollback on processing failure so lastGetBatchMessage is cleared.

Example fix

// before
List<CommonMessage> batch = consumer.getMessage(timeout, unit);
process(batch);
consumer.getMessage(...); // throws

// after
List<CommonMessage> batch = consumer.getMessage(timeout, unit);
try {
    process(batch);
    consumer.ack();
} catch (RuntimeException e) {
    consumer.rollback();
    throw e;
}

if (this.lastGetBatchMessage != null) {
    rollback(); // clear the outstanding batch first
}

boolean readyForNextBatch() {
    return this.lastGetBatchMessage == null;
}

When it happens

Trigger: getMessage() is called while this.lastGetBatchMessage != null — a prior batch is still outstanding because neither ack() nor rollback() has cleared it. Multi-threaded or re-entrant getMessage calls trip this immediately.

Common situations: Processing a batch in a separate thread while the main loop calls getMessage again; an exception between getMessage and ack that skips rollback; forgetting to ack after successful processing.

Related errors

• mq get/ack not support concurrent & async ack
• Failed to fetch the data after: {}
• mq get/ack not support concurrent & async ack
• Receive pulsar batch message error
• error

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