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 RocketMQCanalConnector.getListWithoutAck (the Message, non-flat variant) when lastGetBatchMessage is non-null — i.e. a prior batch was fetched but not yet acked or rolled back. The class Javadoc at the top of the file explicitly states get and ack must be strictly serialized on one thread for the MQ connector, unlike the TCP SimpleCanalConnector.

Source

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

    public void unsubscribe() throws CanalClientException {
        this.rocketMQConsumer.unsubscribe(this.topic);
    }

    @Override
    public List<Message> getList(Long timeout, TimeUnit unit) throws CanalClientException {
        List<Message> messages = getListWithoutAck(timeout, unit);
        if (messages != null && !messages.isEmpty()) {
            ack();
        }
        return messages;
    }

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

            ConsumerBatchMessage 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 Lists.newArrayList();
    }

    @Override
    public List<FlatMessage> getFlatList(Long timeout, TimeUnit unit) throws CanalClientException {
        List<FlatMessage> messages = getFlatListWithoutAck(timeout, unit);
        if (messages != null && !messages.isEmpty()) {

View on GitHub (pinned to 87be50e876)

Solutions

1. Always pair getListWithoutAck with ack() or rollback() before the next fetch, using try/finally.
2. Prefer getList() which auto-acks non-empty batches.
3. Keep the connector single-threaded for get/ack as the class Javadoc mandates.
4. On processing failure, call rollback() to clear lastGetBatchMessage.

Example fix

// before
List<Message> msgs = connector.getListWithoutAck(1, TimeUnit.SECONDS);
doWork(msgs); // throws -> ack skipped
connector.ack();
// after
List<Message> msgs = connector.getListWithoutAck(1, TimeUnit.SECONDS);
try {
    doWork(msgs);
    connector.ack();
} catch (Exception e) {
    connector.rollback();
    throw e;
}

private boolean lastBatchPending = false;
List<Message> safeGetListWithoutAck(CanalMQConnector c, long t, TimeUnit u) throws CanalClientException {
    if (lastBatchPending) throw new IllegalStateException("previous batch not acked/rolled back");
    List<Message> msgs = c.getListWithoutAck(t, u);
    lastBatchPending = (msgs != null && !msgs.isEmpty());
    return msgs;
}

List<Message> msgs = connector.getListWithoutAck(1, TimeUnit.SECONDS);
try {
    process(msgs);
    connector.ack();
} catch (RuntimeException e) {
    connector.rollback();
    throw e;
}

When it happens

Trigger: Calling getListWithoutAck() twice in succession without an intervening ack()/rollback(); a processing exception that bypasses ack; concurrent threads sharing the same connector; using getWithoutAck(batchSize) which is unsupported then falling back without resetting state.

Common situations: Porting a TCP-mode consumer that relied on batchId-based ack to RocketMQ mode; exception swallowing in the processing loop that skips ack; multi-threaded consumer designs that parallelize get across worker threads.

Related errors

• Start RocketMQ consumer error
• Failed to fetch the data after: {timeout}
• mq not support this method
• mq get/ack not support concurrent & async ack
• pattern tag cannot multi: {}

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