{"record":{"id":"ea8cb9671af85477","repo":"apache/seatunnel","slug":"timed-out-waiting-for-error-sink-queue-to-drain-j","errorCode":null,"errorMessage":"Timed out waiting for error sink queue to drain. jobId=%d, pluginName=%s, pendingRows=%d","messagePattern":"Timed out waiting for error sink queue to drain\\. jobId=(.+?), pluginName=(.+?), pendingRows=(.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/error/DefaultErrorSinkWriter.java","lineNumber":640,"sourceCode":"                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()));\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        }","sourceCodeStart":622,"sourceCodeEnd":658,"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#L622-L658","documentation":"RuntimeException thrown by waitForPendingRows (called from flushInternal) when the error sink queue still has pending rows after the configured drain timeout. It polls pendingRows every 10ms up to a deadline, rechecking worker failure; on timeout it reports jobId, plugin and remaining pending row count, indicating the error sink could not keep up or the worker is stuck.","triggerScenarios":"flush() is invoked (checkpoint or close) and pendingRows.get() > 0 when System.currentTimeMillis() passes the deadline; each poll first rethrows any worker failure, so this only fires when the worker is alive but too slow or blocked.","commonSituations":"Huge dirty-row burst just before checkpoint; error sink writing to a slow/hung remote system; drain timeout configured too small for the sink throughput.","solutions":["Increase the error-sink flush/drain timeout configuration","Reduce dirty-row volume or increase error sink parallelism/throughput","Investigate the error sink's downstream latency (network, target DB)","Enable error-sink metrics/logging to confirm the worker is progressing, not hung"],"exampleFix":"# before\nerror-sink {\n  ErrorQueueCapacity = 100\n  flush_interval = 1000\n}\n# after\nerror-sink {\n  ErrorQueueCapacity = 10000\n  flush_interval = 10000\n  drain_timeout = 60000\n}","handlingStrategy":"retry","validationCode":"if (pendingRows.get() > errorQueueCapacity * 0.9) {\n    LOG.warn(\"Error sink near capacity; consider flushing early\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    writer.flush();\n} catch (RuntimeException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"Timed out waiting\")) {\n        LOG.warn(\"Error sink drain timeout; retrying after worker recovery\");\n        // retry flush or fail checkpoint to trigger restart\n    }\n    throw e;\n}","preventionTips":["Tune drain timeout to exceed worst-case error sink latency","Prevent huge dirty-row bursts immediately before checkpoints","Monitor error sink throughput and lag","Use a faster local error sink if the remote one is slow"],"tags":["error-sink","timeout","backpressure","flush","zeta-engine"],"backgroundTag":"request-timeout","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"}