{"record":{"id":"7c08b1f93ecbfabb","repo":"alibaba/canal","slug":"mq-get-ack-not-support-concurrent-async-ack-7c08b1","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":"connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/consumer/CanalRabbitMQConsumer.java","lineNumber":166,"sourceCode":"            logger.error(\"Put message to queue error\", e);\n            throw new RuntimeException(e);\n        }\n        boolean isCompleted;\n        try {\n            isCompleted = batchMessage.waitFinish(batchProcessTimeout);\n        } catch (InterruptedException e) {\n            logger.error(\"Interrupted when waiting messages to be finished.\", e);\n            throw new RuntimeException(e);\n        }\n        boolean isSuccess = batchMessage.isSuccess();\n        return isCompleted && isSuccess;\n    }\n\n    @Override\n    public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {\n        try {\n            if (this.lastGetBatchMessage != null) {\n                throw new CanalClientException(\"mq get/ack not support concurrent & async ack\");\n            }\n\n            ConsumerBatchMessage<CommonMessage> 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 null;\n    }\n\n    @Override\n    public void rollback() {\n        try {\n            if (this.lastGetBatchMessage != null) {","sourceCodeStart":148,"sourceCodeEnd":184,"githubUrl":"https://github.com/alibaba/canal/blob/87be50e87686a3e8af08c368d0e1ffd1f59eb04a/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/consumer/CanalRabbitMQConsumer.java#L148-L184","documentation":"Thrown by CanalRabbitMQConsumer.getMessage when a previous batch (lastGetBatchMessage) has not yet been acknowledged or rolled back. The connector enforces strict get-then-ack ordering: you must ack or rollback the current batch before requesting the next one, because it does not support concurrent or async acknowledgement.","triggerScenarios":"getMessage() is called while this.lastGetBatchMessage != null — i.e. a prior getMessage returned a batch and neither ack nor rollback has cleared the field. Re-entrant or multi-threaded calls to getMessage trip this immediately.","commonSituations":"Calling getMessage from multiple threads; forgetting to call ack()/rollback() after processing a batch; an exception in processing that skips the ack and the caller loops back to getMessage.","solutions":["Ensure every getMessage is followed by exactly one ack() or rollback() before the next getMessage call.","Serialize getMessage/ack/rollback on a single thread or with an external lock — the connector is not concurrency-safe across these.","Wrap batch processing in try/finally so rollback() runs if processing throws, clearing lastGetBatchMessage."],"exampleFix":"// before\nList<CommonMessage> batch = consumer.getMessage(timeout, unit);\nprocess(batch);            // if this throws, next getMessage fails\ndoSomethingElse();\nconsumer.getMessage(...); // throws 'mq get/ack not support concurrent & async ack'\n\n// after — ack or rollback before the next get\nList<CommonMessage> batch = consumer.getMessage(timeout, unit);\ntry {\n    process(batch);\n    consumer.ack();\n} catch (Exception e) {\n    consumer.rollback();\n    throw e;\n}","handlingStrategy":"validation","validationCode":"if (this.lastGetBatchMessage != null) {\n    // ack or rollback the previous batch before getting the next\n    rollback();\n}","typeGuard":"boolean readyForNextBatch() {\n    return this.lastGetBatchMessage == null;\n}","tryCatchPattern":null,"preventionTips":["Always ack() or rollback() after getMessage() before the next get.","Serialize getMessage/ack/rollback — the connector is not thread-safe.","Use try/finally to rollback on processing failure."],"tags":["rabbitmq","consumer","concurrency","canal-connector","acknowledgement"],"backgroundTag":null,"analyzedSha":"87be50e87686a3e8af08c368d0e1ffd1f59eb04a","analyzedAt":"2026-08-14T04:30:11.918Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}