{"record":{"id":"2368eda1a6a19b15","repo":"alibaba/canal","slug":"mq-get-ack-not-support-concurrent-async-ack-2368ed","errorCode":null,"errorMessage":"mq get/ack not support concurrent & async ack","messagePattern":"mq get/ack not support concurrent & async ack","errorType":"exception","errorClass":"CanalClientException","httpStatus":null,"severity":"error","filePath":"client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java","lineNumber":232,"sourceCode":"\n    public void unsubscribe() throws CanalClientException {\n        this.rocketMQConsumer.unsubscribe(this.topic);\n    }\n\n    @Override\n    public List<Message> getList(Long timeout, TimeUnit unit) throws CanalClientException {\n        List<Message> messages = getListWithoutAck(timeout, unit);\n        if (messages != null && !messages.isEmpty()) {\n            ack();\n        }\n        return messages;\n    }\n\n    @Override\n    public List<Message> getListWithoutAck(Long timeout, TimeUnit unit) throws CanalClientException {\n        try {\n            if (this.lastGetBatchMessage != null) {\n                throw new CanalClientException(\"mq get/ack not support concurrent & async ack\");\n            }\n\n            ConsumerBatchMessage batchMessage = messageBlockingQueue.poll(timeout, unit);\n            if (batchMessage != null) {\n                this.lastGetBatchMessage = batchMessage;\n                return batchMessage.getData();\n            }\n        } catch (InterruptedException ex) {\n            logger.warn(\"Get message timeout\", ex);\n            throw new CanalClientException(\"Failed to fetch the data after: \" + timeout);\n        }\n        return Lists.newArrayList();\n    }\n\n    @Override\n    public List<FlatMessage> getFlatList(Long timeout, TimeUnit unit) throws CanalClientException {\n        List<FlatMessage> messages = getFlatListWithoutAck(timeout, unit);\n        if (messages != null && !messages.isEmpty()) {","sourceCodeStart":214,"sourceCodeEnd":250,"githubUrl":"https://github.com/alibaba/canal/blob/87be50e87686a3e8af08c368d0e1ffd1f59eb04a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java#L214-L250","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Always pair getListWithoutAck with ack() or rollback() before the next fetch, using try/finally.","Prefer getList() which auto-acks non-empty batches.","Keep the connector single-threaded for get/ack as the class Javadoc mandates.","On processing failure, call rollback() to clear lastGetBatchMessage."],"exampleFix":"// before\nList<Message> msgs = connector.getListWithoutAck(1, TimeUnit.SECONDS);\ndoWork(msgs); // throws -> ack skipped\nconnector.ack();\n// after\nList<Message> msgs = connector.getListWithoutAck(1, TimeUnit.SECONDS);\ntry {\n    doWork(msgs);\n    connector.ack();\n} catch (Exception e) {\n    connector.rollback();\n    throw e;\n}","handlingStrategy":"validation","validationCode":"private boolean lastBatchPending = false;\nList<Message> safeGetListWithoutAck(CanalMQConnector c, long t, TimeUnit u) throws CanalClientException {\n    if (lastBatchPending) throw new IllegalStateException(\"previous batch not acked/rolled back\");\n    List<Message> msgs = c.getListWithoutAck(t, u);\n    lastBatchPending = (msgs != null && !msgs.isEmpty());\n    return msgs;\n}","typeGuard":null,"tryCatchPattern":"List<Message> msgs = connector.getListWithoutAck(1, TimeUnit.SECONDS);\ntry {\n    process(msgs);\n    connector.ack();\n} catch (RuntimeException e) {\n    connector.rollback();\n    throw e;\n}","preventionTips":["Pair every getListWithoutAck with ack()/rollback() in try/finally.","Prefer getList() which auto-acks non-empty batches.","Keep get/ack on one thread per the class Javadoc.","When porting from TCP mode, remove batchId-based ack and use parameterless ack()."],"tags":["rocketmq","canal-client","concurrency","ack","state"],"backgroundTag":null,"analyzedSha":"87be50e87686a3e8af08c368d0e1ffd1f59eb04a","analyzedAt":"2026-08-14T04:30:11.918Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}