{"record":{"id":"83c80c669c99e4f5","repo":"alibaba/canal","slug":"clientid-s-has-last-batch-s-isn-t-ack-maybe","errorCode":null,"errorMessage":"clientId:%s has last batch:[%s] isn't ack , maybe loss data","messagePattern":"clientId:(.+?) has last batch:\\[(.+?)\\] isn't ack , maybe loss data","errorType":"exception","errorClass":"CanalServerException","httpStatus":null,"severity":"error","filePath":"server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java","lineNumber":251,"sourceCode":"     * b. 如果timeout不为null\n     *    1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回\n     *    2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少\n     * \n     * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)\n     * </pre>\n     */\n    @Override\n    public Message get(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)\n                                                                                                 throws CanalServerException {\n        checkStart(clientIdentity.getDestination());\n        checkSubscribe(clientIdentity);\n        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());\n        synchronized (canalInstance) {\n            // 获取到流式数据中的最后一批获取的位置\n            PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().getLastestBatch(clientIdentity);\n\n            if (positionRanges != null) {\n                throw new CanalServerException(String.format(\"clientId:%s has last batch:[%s] isn't ack , maybe loss data\",\n                    clientIdentity.getClientId(),\n                    positionRanges));\n            }\n\n            Events<Event> events = null;\n            Position start = canalInstance.getMetaManager().getCursor(clientIdentity);\n            events = getEvents(canalInstance.getEventStore(), start, batchSize, timeout, unit);\n\n            if (CollectionUtils.isEmpty(events.getEvents())) {\n                logger.debug(\"get successfully, clientId:{} batchSize:{} but result is null\",\n                    clientIdentity.getClientId(),\n                    batchSize);\n                return new Message(-1, true, new ArrayList()); // 返回空包，避免生成batchId，浪费性能\n            } else {\n                // 记录到流式信息\n                Long batchId = canalInstance.getMetaManager().addBatch(clientIdentity, events.getPositionRange());\n                boolean raw = isRaw(canalInstance.getEventStore());\n                List entrys = null;","sourceCodeStart":233,"sourceCodeEnd":269,"githubUrl":"https://github.com/alibaba/canal/blob/87be50e87686a3e8af08c368d0e1ffd1f59eb04a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java#L233-L269","documentation":"Thrown by CanalServerWithEmbedded.get() when MetaManager.getLastestBatch(clientIdentity) returns a non-null PositionRange. A non-null result means the client already holds an outstanding previously-fetched batch that was never acknowledged (ack) or rolled back (rollback). To prevent data loss the server refuses to hand out a new batch until the previous one is resolved.","triggerScenarios":"Calling embeddedServer.get(clientIdentity, batchSize, timeout, unit) after a prior get() whose returned Message was neither acked nor rolled back. The MetaManager (memory or zookeeper backed) still records the last batch as pending.","commonSituations":"Client crashed or was killed mid-batch without acking; client bug that consumes get() but only conditionally calls ack(); long processing time where the prior batch is still in-flight; restart of the client while the server-side batch metadata persists in ZooKeeper.","solutions":["Ack or rollback the previously delivered batch (use the batchId from the last Message) before calling get() again.","On client startup, call rollback(clientIdentity) to clear any stale outstanding batch left from a prior run.","If the batch is legitimately lost, call rollback to reset the cursor to the last acked position, then resume get().","Review client code to guarantee ack() is always invoked in a finally block after successful processing."],"exampleFix":"// before: fetch without resolving prior batch\nMessage msg = server.get(clientId, 1000, 1L, TimeUnit.SECONDS);\nprocess(msg);\nserver.ack(clientId, msg.getId());\n// after: clear stale batch on connect, ack in finally\nserver.rollback(clientId); // clear any pending batch at startup\nMessage msg = server.get(clientId, 1000, 1L, TimeUnit.SECONDS);\ntry {\n    process(msg);\n    server.ack(clientId, msg.getId());\n} catch (Exception e) {\n    server.rollback(clientId, msg.getId());\n}","handlingStrategy":"validation","validationCode":"// before calling get(), ensure no outstanding batch is pending\nPositionRange last = metaManager.getLastestBatch(clientIdentity);\nif (last != null) {\n    server.rollback(clientIdentity); // clear the stale batch first\n}","typeGuard":null,"tryCatchPattern":"try {\n    Message msg = server.get(clientId, batchSize, timeout, unit);\n} catch (CanalServerException e) {\n    if (e.getMessage().contains(\"isn't ack\")) {\n        server.rollback(clientId); // resolve the pending batch then retry once\n        return server.get(clientId, batchSize, timeout, unit);\n    }\n    throw e;\n}","preventionTips":["On client startup always rollback to clear any leftover batch from a prior crash.","Wrap processing in try/finally so ack() always runs on success and rollback() on failure.","Use a single thread per clientId to serialize get/ack and avoid interleaving."],"tags":["embedded-server","batch-processing","ack","data-loss-prevention"],"backgroundTag":null,"analyzedSha":"87be50e87686a3e8af08c368d0e1ffd1f59eb04a","analyzedAt":"2026-08-14T04:30:11.918Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}