alibaba/canal · error · CanalClientException

Failed to fetch the data after: {timeout}

Error message

Failed to fetch the data after: {timeout}

What it means

Thrown by RocketMQCanalConnector.getListWithoutAck when the blocking queue poll is interrupted (InterruptedException) while waiting for a batch. The connector converts interruption into a CanalClientException naming the timeout. This indicates thread interruption, not a benign empty-poll timeout (which returns an empty list).

Source

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

        }
        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()) {
            ack();
        }
        return messages;
    }

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

View on GitHub (pinned to 87be50e876)

Solutions

1. Detect expected shutdown interruptions and exit the loop gracefully.
2. Restore interrupt flag in the catch and propagate only if unexpected.
3. Ensure disconnect happens after the consumer loop terminates, not concurrently with poll.
4. Avoid interrupting the consumer thread for non-shutdown reasons.

Example fix

// before
List<Message> msgs = connector.getListWithoutAck(1, TimeUnit.SECONDS);
// after
List<Message> msgs;
try {
    msgs = connector.getListWithoutAck(1, TimeUnit.SECONDS);
} catch (CanalClientException e) {
    if (shuttingDown) break;
    Thread.currentThread().interrupt();
    throw e;
}

if (shuttingDown) return; // skip polling during shutdown

try {
    msgs = connector.getListWithoutAck(1, TimeUnit.SECONDS);
} catch (CanalClientException e) {
    if (shuttingDown && Thread.currentThread().isInterrupted()) {
        logger.info("Shutting down consumer; exiting loop.");
        return;
    }
    Thread.currentThread().interrupt();
    throw e;
}

When it happens

Trigger: Thread.interrupt() invoked on the consumer thread during poll; executor shutdown; container shutdown that interrupts worker threads; calling disconnect concurrently.

Common situations: Application shutdown interrupting the consumer thread; running inside a managed thread pool that cancels tasks; tests that time out and interrupt the consumer.

Understand the failure class

• Timeouts: ETIMEDOUT, deadlines, and hung requests — what actually expires when a request times out.

Related errors

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

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