alibaba/canal · warning · CanalClientException

Failed to fetch the data after: {}

Error message

Failed to fetch the data after: {}

What it means

Thrown by CanalRocketMQConsumer.getMessage when messageBlockingQueue.poll(timeout, unit) is interrupted while waiting for a batch. This is a thread-interruption signal (usually shutdown), not a broker error or a true poll timeout — the connector reuses the 'Failed to fetch' wording but the cause is InterruptedException.

Source

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

        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) {
                this.lastGetBatchMessage.fail();
            }
        } finally {
            this.lastGetBatchMessage = null;
        }
    }

    @Override
    public void ack() {
        try {

View on GitHub (pinned to 87be50e876)

Solutions

1. During shutdown, log and exit, restoring the interrupt flag.
2. If unexpected, locate and stop the source of Thread.interrupt() on the consumer thread.
3. Do not confuse this with a broker timeout — increasing poll timeout will not prevent an interrupt.

Example fix

// before
} catch (InterruptedException ex) {
    logger.warn("Get message timeout", ex);
    throw new CanalClientException("Failed to fetch the data after: " + timeout);
}

// after — accurate interruption handling
} catch (InterruptedException ex) {
    logger.warn("Get message interrupted", ex);
    Thread.currentThread().interrupt();
    throw new CanalClientException("Get message interrupted after: " + timeout);
}

} catch (InterruptedException ex) {
    logger.warn("Get message interrupted", ex);
    Thread.currentThread().interrupt();
    if (isShuttingDown()) return Collections.emptyList();
    throw new CanalClientException("Get message interrupted after: " + timeout);
}

When it happens

Trigger: poll(timeout, unit) at line 192 throws InterruptedException because Thread.interrupt() was invoked on the consuming thread — e.g. during executor shutdownNow() or application stop.

Common situations: Controlled shutdown interrupting the consumer thread; scheduler/executor shutdownNow; an explicit interrupt from upstream code. Benign during a clean stop.

Related errors

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

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