alibaba/canal · error · CanalParseException

consume failed!

Error message

consume failed!

What it means

Thrown by the EventTransactionBuffer flush callback inside AbstractEventParser's constructor when consumeTheEventAndProfilingIfNecessary(transaction) returns false for a transaction while the parser is still running. It indicates the sink/store rejected the transaction (the consume/sink pipeline returned a non-success result), so parsing cannot safely continue.

Source

Thrown at parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java:134

    protected void afterDump(ErosaConnection connection) {
    }

    public void sendAlarm(String destination, String msg) {
        if (this.alarmHandler != null) {
            this.alarmHandler.sendAlarm(destination, msg);
        }
    }

    public AbstractEventParser(){
        // 初始化一下
        transactionBuffer = new EventTransactionBuffer(transaction -> {
            boolean successed = consumeTheEventAndProfilingIfNecessary(transaction);
            if (!running) {
                return;
            }

            if (!successed) {
                throw new CanalParseException("consume failed!");
            }

            LogPosition position = buildLastTransactionPosition(transaction);
            if (position != null) { // 可能position为空
                logPositionManager.persistLogPosition(AbstractEventParser.this.destination, position);
            }
        });
    }

    public void start() {
        super.start();
        MDC.put("destination", destination);
        // 配置transaction buffer
        // 初始化缓冲队列
        transactionBuffer.setBufferSize(transactionSize);// 设置buffer大小
        transactionBuffer.start();
        // 构造bin log parser
        binlogParser = buildParser();// 初始化一下BinLogParser

View on GitHub (pinned to 87be50e876)

Solutions

1. Increase the event store capacity (canal.instance.memory.buffer.size / buffer.retain) so the sink is not rejected.
2. Check the downstream consumer is draining batches fast enough; a stalled client fills the store.
3. Inspect logs just before this error for sink/store rejection messages to find the true sink failure.
4. If using a custom EventSink, ensure it returns true on success and only false intentionally; do not swallow exceptions as false.

// parser-level; wrap the sink so consume failure is observable
try { transactionBuffer.flush(); }
catch (CanalParseException e) {
    if ("consume failed!".equals(e.getMessage())) { throttleOrScaleStore(); throw e; }
    throw e;
}

When it happens

Trigger: The transaction buffer flushes a transaction and invokes the callback; consumeTheEventAndProfilingIfNecessary returns false. This happens when the downstream CanalEventStore.doSink or the sink chain signals failure (store full/rejected, sink filter rejected all, or sink threw). With running still true, the parser raises CanalParseException 'consume failed!'.

Common situations: Event store (MemoryEventStore) is full and sink is rejected repeatedly, a custom EventSink returns false, the sink throws and is translated to false, or the destination's store capacity is too small for the transaction burst. Often surfaces during high-throughput bursts or when a slow downstream backs up the store.

Related errors

• can't find start position for {}
• can't find init table meta for {} with position : {}
• dump address %s has an error, retrying.
• disconnect address %s has an error, retrying.
• unknow column : + name

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