{"record":{"id":"14d5c74bf3e106ac","repo":"apache/seatunnel","slug":"interrupted-while-waiting-for-error-sink-queue","errorCode":null,"errorMessage":"Interrupted while waiting for error sink queue","messagePattern":"Interrupted while waiting for error sink queue","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"warning","filePath":"seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/error/DefaultErrorSinkWriter.java","lineNumber":649,"sourceCode":"    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()));\n            }\n            try {\n                Thread.sleep(10L);\n            } catch (InterruptedException e) {\n                Thread.currentThread().interrupt();\n                throw new RuntimeException(\"Interrupted while waiting for error sink queue\", e);\n            }\n        }\n    }\n\n    private void throwWorkerFailureIfAny() throws Exception {\n        Throwable failure = workerFailure;\n        if (failure == null) {\n            return;\n        }\n        if (failure instanceof Error) {\n            throw (Error) failure;\n        }\n        if (failure instanceof Exception) {\n            throw (Exception) failure;\n        }\n        throw new RuntimeException(failure);\n    }\n","sourceCodeStart":631,"sourceCodeEnd":667,"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#L631-L667","documentation":"RuntimeException thrown by waitForPendingRows when its Thread.sleep(10ms) poll loop is interrupted while waiting for the error sink queue to drain during flush. The interrupt flag is restored and the exception is rethrown wrapped, aborting the flush operation.","triggerScenarios":"flushInternal -> waitForPendingRows is sleeping between pendingRows polls and the thread is interrupted - normally by task cancellation, checkpoint timeout, or engine shutdown while a flush is in progress.","commonSituations":"Cancelling a job during checkpoint flush; checkpoint timeout killing a slow flush; engine worker shutdown while error rows are still pending.","solutions":["Expected during cancellation - no action if the job was intentionally stopped","Raise checkpoint timeout if checkpoints are being cancelled during heavy flush","Speed up the error sink or drain it earlier so flush waits are short","Check for external code interrupting task threads unexpectedly"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    writer.flush();\n} catch (RuntimeException e) {\n    if (e.getCause() instanceof InterruptedException) {\n        Thread.currentThread().interrupt();\n        LOG.warn(\"Flush interrupted while draining error sink queue\");\n    }\n    throw e;\n}","preventionTips":["Treat interrupts during flush as expected during cancellation","Increase checkpoint timeout to reduce flush interruption","Drain the error queue incrementally instead of one long flush","Avoid interrupting task threads from custom code"],"tags":["error-sink","interrupted-exception","flush","cancellation","zeta-engine"],"backgroundTag":"task-cancelled","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"}