{"record":{"id":"7ce1372ba227609f","repo":"apache/seatunnel","slug":"interrupted-while-waiting-for-error-sink-worker-to-7ce137","errorCode":null,"errorMessage":"Interrupted while waiting for error sink worker to close after interrupt","messagePattern":"Interrupted while waiting for error sink worker to close after interrupt","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/error/DefaultErrorSinkWriter.java","lineNumber":625,"sourceCode":"            return;\n        }\n        try {\n            workerThread.join(timeoutMillis);\n        } catch (InterruptedException e) {\n            Thread.currentThread().interrupt();\n            log.warn(\"Interrupted while waiting for error sink worker to close\");\n        }\n\n        if (workerThread.isAlive()) {\n            log.warn(\n                    \"Error sink worker thread did not terminate within {} ms, interrupting it\",\n                    timeoutMillis);\n            workerThread.interrupt();\n            try {\n                workerThread.join(Math.min(5_000L, timeoutMillis));\n            } catch (InterruptedException e) {\n                Thread.currentThread().interrupt();\n                log.warn(\n                        \"Interrupted while waiting for error sink worker to close after interrupt\");\n            }\n        }\n    }\n\n    private void waitForPendingRows(long timeoutMillis) throws Exception {\n        AtomicInteger currentPendingRows = this.pendingRows;\n        if (currentPendingRows == null) {\n            return;\n        }\n        long deadline = System.currentTimeMillis() + timeoutMillis;\n        while (currentPendingRows.get() > 0) {\n            throwWorkerFailureIfAny();\n            if (System.currentTimeMillis() >= deadline) {\n                throw new RuntimeException(\n                        String.format(\n                                \"Timed out waiting for error sink queue to drain. jobId=%d, pluginName=%s, pendingRows=%d\",\n                                jobId, sinkConfig.getPluginName(), currentPendingRows.get()));","sourceCodeStart":607,"sourceCodeEnd":643,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/error/DefaultErrorSinkWriter.java#L607-L643","documentation":"After the first join times out, close() interrupts the error-sink worker thread and joins again for up to 5 seconds. If that second join is itself interrupted, the interrupt flag is restored and this warning is logged. This is a nested-shutdown diagnostics warning indicating heavy contention on the closing thread's interrupt status.","triggerScenarios":"A second interrupt arrives on the closing thread while it waits in workerThread.join(min(5000, timeoutMillis)) after having interrupted the worker — typically aggressive task cancellation or engine shutdown racing sink close.","commonSituations":"Job kill/failover storm interrupting all task threads repeatedly; engine shutdown hooks firing while close() is still in progress.","solutions":["Treat as benign if the job was being cancelled or the engine shut down.","If recurring in normal operation, audit code that interrupts task threads (retry/cancel logic) to avoid double interrupts.","Ensure error-sink worker finishes quickly so close() doesn't need the interrupt path at all."],"exampleFix":"null","handlingStrategy":"try-catch","validationCode":"null","typeGuard":"null","tryCatchPattern":"null","preventionTips":["Avoid aggressive repeated interrupts during shutdown","Let close() run to completion before cancelling tasks"],"tags":["threading","interruption","shutdown"],"backgroundTag":"thread-interrupted","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"}