{"record":{"id":"d9f7aff1a2539aef","repo":"alibaba/canal","slug":"state-is-not-correct-in-non-transaction","errorCode":null,"errorMessage":"state is not correct in non-transaction","messagePattern":"state is not correct in non-transaction","errorType":"exception","errorClass":"CanalSinkException","httpStatus":null,"severity":"critical","filePath":"sink/src/main/java/com/alibaba/otter/canal/sink/entry/group/TimelineTransactionBarrier.java","lineNumber":66,"sourceCode":"            super.await(event, timeout, unit);\n        } catch (InterruptedException e) {\n            // 出现线程中断，可能是因为关闭或者主备切换\n            // 主备切换对应的事务尾会未正常发送，需要强制设置为事务结束，允许其他队列通过\n            reset();\n            throw e;\n        }\n    }\n\n    public void clear(Event event) {\n        super.clear(event);\n\n        // 应该先判断2，再判断是否是事务尾，因为事务尾也可以导致txState的状态为2\n        // 如果先判断事务尾，那么2的状态可能永远没机会被修改了，系统出现死锁\n        // CanalSinkException被注释的代码是不是可以放开？？我们内部使用的时候已经放开了，从代码逻辑的分析上以及实践效果来看，应该抛异常\n        if (txState.intValue() == 2) {// 非事务中\n            boolean result = txState.compareAndSet(2, 0);\n            if (result == false) {\n                throw new CanalSinkException(\"state is not correct in non-transaction\");\n            }\n        } else if (isTransactionEnd(event)) {\n            inTransaction.set(false); // 事务结束并且已经成功写入store，清理标记，进入重新排队判断，允许新的事务进入\n            boolean result = txState.compareAndSet(1, 0);\n            if (result == false) {\n                throw new CanalSinkException(\"state is not correct in transaction\");\n            }\n        }\n    }\n\n    protected boolean isPermit(Event event, long state) {\n        if (txState.intValue() == 1 && inTransaction.get()) { // 如果处于事务中，直接允许通过。因为事务头已经做过判断\n            return true;\n        } else if (txState.intValue() == 0) {\n            boolean result = super.isPermit(event, state);\n            if (result) {\n                // 可能第一条送过来的数据不为Begin，需要做判断处理，如果非事务，允许直接通过，比如DDL语句\n                if (isTransactionBegin(event)) {","sourceCodeStart":48,"sourceCodeEnd":84,"githubUrl":"https://github.com/alibaba/canal/blob/87be50e87686a3e8af08c368d0e1ffd1f59eb04a/sink/src/main/java/com/alibaba/otter/canal/sink/entry/group/TimelineTransactionBarrier.java#L48-L84","documentation":"Thrown in TimelineTransactionBarrier.clear(event) when txState is 2 (non-transaction data in flight) but the CAS txState.compareAndSet(2, 0) fails — meaning another thread already mutated txState between the read and the CAS. This indicates a concurrent state corruption: clear() was invoked for a non-transactional event while another path changed the state, breaking the single-writer assumption of the barrier state machine.","triggerScenarios":"Two sink threads (group members) calling clear() concurrently on non-transactional events such that both read txState==2 but only one wins the CAS; an interrupt/reset racing with clear(); misuse of the barrier from multiple threads without the intended group coordination.","commonSituations":"Misconfigured group sink with overlapping threads writing the same barrier; a parser/restarter thread interrupting (reset()) during clear(); custom code that invokes the barrier outside the normal await->store->clear sequence.","solutions":["Ensure each group sink thread follows the strict await(event) -> store -> clear(event) sequence without skipping steps.","Avoid sharing a single TimelineTransactionBarrier across more threads than the configured group size.","On this exception, treat the pipeline as poisoned: reset the instance (stop/start) to restore txState to 0 rather than continuing.","Upgrade canal version if hitting this during master-slave switch, as the reset-on-interrupt path was added to handle exactly that race."],"exampleFix":null,"handlingStrategy":"fallback","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    barrier.clear(event);\n} catch (CanalSinkException e) {\n    if (e.getMessage().contains(\"non-transaction\")) {\n        logger.error(\"barrier state corrupted, resetting instance\", e);\n        canalInstance.stop();\n        canalInstance.start(); // re-stream from last good cursor\n    } else {\n        throw e;\n    }\n}","preventionTips":["Run exactly one sink thread per group slot to honor the barrier's single-writer design.","Always pair await(event) with clear(event); never skip clear for a permitted event.","Keep canal version current to get interrupt/reset race fixes."],"tags":["sink","transaction","concurrency","state-machine","group-sink"],"backgroundTag":null,"analyzedSha":"87be50e87686a3e8af08c368d0e1ffd1f59eb04a","analyzedAt":"2026-08-14T04:30:11.918Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}