{"record":{"id":"edcaeaf61e6422d8","repo":"apache/seatunnel","slug":"unsafe-invoke-the-current-thread-s-has-not-acqu","errorCode":null,"errorMessage":"Unsafe invoke, the current thread[%s] has not acquired the lock[%s].","messagePattern":"Unsafe invoke, the current thread\\[(.+?)\\] has not acquired the lock\\[(.+?)\\]\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/reader/fetcher/SplitFetcher.java","lineNumber":183,"sourceCode":"                    String.format(\n                            \"SplitFetcher thread %d received unexpected exception while polling the records\",\n                            fetcherId),\n                    e);\n        }\n\n        // re-acquire lock as all post-processing steps, need it\n        lock.lock();\n        try {\n            this.runningTask = null;\n        } finally {\n            lock.unlock();\n        }\n        return true;\n    }\n\n    private SplitFetcherTask getNextTaskUnsafe() {\n        if (!lock.isHeldByCurrentThread()) {\n            throw new RuntimeException(\n                    String.format(\n                            \"Unsafe invoke, the current thread[%s] has not acquired the lock[%s].\",\n                            Thread.currentThread().getName(), this.lock.toString()));\n        }\n\n        try {\n            if (!taskQueue.isEmpty()) {\n                // execute tasks in taskQueue first\n                return taskQueue.poll();\n            } else if (!assignedSplits.isEmpty()) {\n                // use fallback task = fetch if there is at least one split\n                return fetchTask;\n            } else {\n                // nothing to do, wait for signal\n                nonEmpty.await();\n                return taskQueue.poll();\n            }\n        } catch (InterruptedException e) {","sourceCodeStart":165,"sourceCodeEnd":201,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/reader/fetcher/SplitFetcher.java#L165-L201","documentation":"SplitFetcher methods suffixed 'Unsafe' (getNextTaskUnsafe) must only be called while holding the fetcher's ReentrantLock. getNextTaskUnsafe explicitly checks lock.isHeldByCurrentThread() and throws if the calling thread does not own the lock, guarding internal task-queue state from unsynchronized access.","triggerScenarios":"Any code path calling getNextTaskUnsafe (directly or via runOnce refactoring) from a thread that has not acquired SplitFetcher's lock — typically new custom code or subclasses touching fetcher internals, since run() normally acquires the lock before runOnce().","commonSituations":"Custom SplitFetcher subclass overriding run()/runOnce() without wrapping calls in lock.lock()/unlock(); external tooling invoking private-adjacent unsafe methods; concurrency refactors that dropped lock acquisition.","solutions":["Acquire the lock before calling: lock.lock() then try { getNextTaskUnsafe(); } finally { lock.unlock(); }","Don't call Unsafe methods from outside SplitFetcher.run() — use public APIs (addTask, shutdown, wakeUp) which handle locking","If subclassing SplitFetcher, mirror the original run() locking structure","Restore the original runOnce()/run() call pattern if a refactor removed lock acquisition"],"exampleFix":"// before\nsplitFetcher.getNextTaskUnsafe(); // wrong thread, no lock\n// after\nsplitFetcher.lock.lock();\ntry {\n    splitFetcher.getNextTaskUnsafe();\n} finally {\n    splitFetcher.lock.unlock();\n}","handlingStrategy":"type-guard","validationCode":"// assert ownership before unsafe calls\nif (!splitFetcher.lock.isHeldByCurrentThread()) {\n    throw new AssertionError(\"must hold SplitFetcher lock before unsafe calls\");\n}","typeGuard":"boolean canCallUnsafe(SplitFetcher f) {\n    return f.lock.isHeldByCurrentThread();\n}","tryCatchPattern":null,"preventionTips":["Never call *Unsafe methods from outside SplitFetcher; use public wrappers (addTask, wakeUp, shutdown)","When subclassing, replicate the original run() lock/unlock structure","Avoid refactors that separate unsafe calls from their lock acquisition","Prefer ReentrantLock.isHeldByCurrentThread() assertions in tests for fetcher internals"],"tags":["concurrency","locking","fetcher","reentrantlock"],"backgroundTag":"invalid-state-transition","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}