{"record":{"id":"bf9663311c2e1d7d","repo":"alibaba/canal","slug":"consume-failed","errorCode":null,"errorMessage":"consume failed!","messagePattern":"consume failed!","errorType":"exception","errorClass":"CanalParseException","httpStatus":null,"severity":"error","filePath":"parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java","lineNumber":134,"sourceCode":"    protected void afterDump(ErosaConnection connection) {\r\n    }\r\n\r\n    public void sendAlarm(String destination, String msg) {\r\n        if (this.alarmHandler != null) {\r\n            this.alarmHandler.sendAlarm(destination, msg);\r\n        }\r\n    }\r\n\r\n    public AbstractEventParser(){\r\n        // 初始化一下\r\n        transactionBuffer = new EventTransactionBuffer(transaction -> {\r\n            boolean successed = consumeTheEventAndProfilingIfNecessary(transaction);\r\n            if (!running) {\r\n                return;\r\n            }\r\n\r\n            if (!successed) {\r\n                throw new CanalParseException(\"consume failed!\");\r\n            }\r\n\r\n            LogPosition position = buildLastTransactionPosition(transaction);\r\n            if (position != null) { // 可能position为空\r\n                logPositionManager.persistLogPosition(AbstractEventParser.this.destination, position);\r\n            }\r\n        });\r\n    }\r\n\r\n    public void start() {\r\n        super.start();\r\n        MDC.put(\"destination\", destination);\r\n        // 配置transaction buffer\r\n        // 初始化缓冲队列\r\n        transactionBuffer.setBufferSize(transactionSize);// 设置buffer大小\r\n        transactionBuffer.start();\r\n        // 构造bin log parser\r\n        binlogParser = buildParser();// 初始化一下BinLogParser\r","sourceCodeStart":116,"sourceCodeEnd":152,"githubUrl":"https://github.com/alibaba/canal/blob/87be50e87686a3e8af08c368d0e1ffd1f59eb04a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java#L116-L152","documentation":"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.","triggerScenarios":"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!'.","commonSituations":"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.","solutions":["Increase the event store capacity (canal.instance.memory.buffer.size / buffer.retain) so the sink is not rejected.","Check the downstream consumer is draining batches fast enough; a stalled client fills the store.","Inspect logs just before this error for sink/store rejection messages to find the true sink failure.","If using a custom EventSink, ensure it returns true on success and only false intentionally; do not swallow exceptions as false."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// parser-level; wrap the sink so consume failure is observable\ntry { transactionBuffer.flush(); }\ncatch (CanalParseException e) {\n    if (\"consume failed!\".equals(e.getMessage())) { throttleOrScaleStore(); throw e; }\n    throw e;\n}","preventionTips":["Size the event store above peak transaction volume.","Keep consumers draining fast to avoid store back-pressure.","Never implement a sink that returns false silently."],"tags":["parse","sink","store-backpressure","transaction-buffer"],"backgroundTag":null,"analyzedSha":"87be50e87686a3e8af08c368d0e1ffd1f59eb04a","analyzedAt":"2026-08-14T04:30:11.918Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}