{"record":{"id":"3780e1f5377df0f6","repo":"apache/seatunnel","slug":"the-split-fetcher-manager-has-closed","errorCode":null,"errorMessage":"The split fetcher manager has closed.","messagePattern":"The split fetcher manager has closed\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/reader/fetcher/SplitFetcherManager.java","lineNumber":98,"sourceCode":"                        // Add the exception to the exception list.\n                        uncaughtFetcherException.get().addSuppressed(throwable);\n                    }\n                };\n        String taskThreadName = Thread.currentThread().getName();\n        this.executors =\n                Executors.newCachedThreadPool(\n                        r -> new Thread(r, \"Source Data Fetcher for \" + taskThreadName));\n    }\n\n    public abstract void addSplits(Collection<SplitT> splitsToAdd);\n\n    protected void startFetcher(SplitFetcher<E, SplitT> fetcher) {\n        executors.submit(fetcher);\n    }\n\n    protected synchronized SplitFetcher<E, SplitT> createSplitFetcher() {\n        if (closed) {\n            throw new IllegalStateException(\"The split fetcher manager has closed.\");\n        }\n        // Create SplitReader.\n        SplitReader<E, SplitT> splitReader = splitReaderFactory.get();\n        int fetcherId = fetcherIdGenerator.getAndIncrement();\n        SplitFetcher<E, SplitT> splitFetcher =\n                new SplitFetcher<>(\n                        fetcherId,\n                        elementsQueue,\n                        splitReader,\n                        errorHandler,\n                        () -> {\n                            fetchers.remove(fetcherId);\n                        },\n                        this.splitFinishedHook);\n        fetchers.put(fetcherId, splitFetcher);\n        return splitFetcher;\n    }\n","sourceCodeStart":80,"sourceCodeEnd":116,"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/SplitFetcherManager.java#L80-L116","documentation":"SplitFetcherManager.createSplitFetcher() is guarded against use after the manager has been closed. When closed==true it throws this IllegalStateException instead of creating a fetcher whose executor no longer exists, keeping shutdown state consistent.","triggerScenarios":"Calling addSplits (or any path that lazily creates a fetcher) after SplitFetcherManager.close() has run — e.g. splits arriving while the reader is being closed, or a race between SplitEnumerator assigning splits and source reader closure.","commonSituations":"Job cancel/fail concurrently with split assignment; checkpoint recovery ordering where a stale reader receives splits; user code closing the source reader then adding splits.","solutions":["Check job lifecycle: ensure addSplits isn't called after close — this is usually a race, so check shutdown ordering logs","Retry-safe pattern: catch IllegalStateException and drop splits if the reader is intentionally closed","Fix the race by synchronizing split assignment against reader shutdown in the enumerator/reader glue","Upgrade SeaTunnel if a known race between close() and addSplits() is the trigger"],"exampleFix":"// before\nsplitFetcherManager.addSplits(splits); // may throw after close\n// after\nif (!splitFetcherManager.isClosed()) { // guard via manager state where exposed\n    splitFetcherManager.addSplits(splits);\n}","handlingStrategy":"validation","validationCode":"// guard against adding splits after close\nif (splitFetcherManager.isClosed()) { // expose/query closed state if available\n    LOG.warn(\"split fetcher manager closed; dropping {} splits\", splits.size());\n    return;\n}\nsplitFetcherManager.addSplits(splits);","typeGuard":"boolean canAcceptSplits(SplitFetcherManager<?, ?> m) {\n    try {\n        java.lang.reflect.Field f = m.getClass().getDeclaredField(\"closed\");\n        f.setAccessible(true);\n        return !f.getBoolean(m);\n    } catch (ReflectiveOperationException e) {\n        return false;\n    }\n}","tryCatchPattern":"try {\n    splitFetcherManager.addSplits(splits);\n} catch (IllegalStateException e) {\n    if (e.getMessage().contains(\"split fetcher manager has closed\")) {\n        LOG.warn(\"manager closed during split assignment; ignoring\", e);\n        return;\n    }\n    throw e;\n}","preventionTips":["Synchronize split assignment against source reader close() in enumerator/reader glue","Don't close the source reader while splits may still be assigned","Check shutdown ordering in custom SourceReader implementations","Treat this error as a lifecycle race signal; log job phase around close/addSplits"],"tags":["lifecycle","shutdown","fetcher","race-condition"],"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-14T05:17:10.506Z"}