{"record":{"id":"3a3057a30263c64a","repo":"apache/seatunnel","slug":"error-queue-overflow-for-stage-s-plugin-s","errorCode":null,"errorMessage":"Error queue overflow for stage [%s], plugin [%s]","messagePattern":"Error queue overflow for stage \\[(.+?)\\], plugin \\[(.+?)\\]","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":173,"sourceCode":"        try {\n            switch (stageConfig.getQueueOverflowPolicy()) {\n                case DROP:\n                    pendingRows.incrementAndGet();\n                    if (!queue.offer(errorRow)) {\n                        pendingRows.decrementAndGet();\n                        return false;\n                    }\n                    break;\n                case BLOCK:\n                    pendingRows.incrementAndGet();\n                    enqueueWithBlockPolicy(ctx, errorRow);\n                    break;\n                case FAIL:\n                default:\n                    pendingRows.incrementAndGet();\n                    if (!queue.offer(errorRow)) {\n                        pendingRows.decrementAndGet();\n                        throw new RuntimeException(\n                                String.format(\n                                        \"Error queue overflow for stage [%s], plugin [%s]\",\n                                        ctx.getStage(), ctx.getPluginName()));\n                    }\n                    break;\n            }\n        } catch (InterruptedException e) {\n            Thread.currentThread().interrupt();\n            pendingRows.decrementAndGet();\n            throw new RuntimeException(\"Interrupted while enqueuing error row for error sink\", e);\n        }\n        return true;\n    }\n\n    private void enqueueWithBlockPolicy(RowErrorContext ctx, SeaTunnelRow errorRow)\n            throws Exception {\n        boolean enqueued = false;\n        try {","sourceCodeStart":155,"sourceCodeEnd":191,"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#L155-L191","documentation":"Thrown when ErrorHandlerMode is FAIL (or unrecognized) and the bounded error queue cannot accept the new error row via its non-blocking offer. The row is rejected and the writer fails fast, signalling that the error queue for this stage/plugin is full and rows would be lost.","triggerScenarios":"ErrorHandlerMode.FAIL/default case in writeAndCheckAccepted: pendingRows incremented, queue.offer(errorRow) returns false because the queue capacity is exhausted (worker too slow or stopped draining), then pendingRows is decremented and this RuntimeException is thrown.","commonSituations":"Burst of bad rows exceeding the configured error queue capacity; error sink worker blocked writing downstream (slow/hung sink); misconfigured small queue size combined with high dirty-data rates.","solutions":["Increase the error sink queue capacity configuration","Switch ErrorHandlerMode to ROUTE with an appropriate blocking/overflow policy if backpressure is acceptable","Investigate why the error sink worker is not draining the queue (downstream sink slowness, prior worker failure)","Reduce the rate of bad rows by fixing upstream data quality or filters"],"exampleFix":"# before\nerror-sink {\n  ErrorHandleMode = FAIL\n  ErrorQueueCapacity = 10\n}\n# after\nerror-sink {\n  ErrorHandleMode = ROUTE\n  ErrorQueueCapacity = 10000\n}","handlingStrategy":"validation","validationCode":"long errorRate = estimateDirtyRowsPerSecond();\nlong capacity = errorQueueCapacity;\nif (errorRate * flushIntervalMs / 1000 > capacity) {\n    throw new IllegalArgumentException(\"ErrorQueueCapacity too small for expected dirty rate\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    writer.write(row);\n} catch (RuntimeException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"Error queue overflow\")) {\n        LOG.warn(\"Error queue full, applying backpressure\");\n        // throttle or switch handle mode\n    }\n}","preventionTips":["Size ErrorQueueCapacity against the expected dirty-data rate","Avoid ErrorHandlerMode=FAIL for high-dirty-volume jobs","Monitor error sink worker lag","Fix upstream data quality to reduce dirty rows"],"tags":["error-sink","queue-overflow","backpressure","zeta-engine"],"backgroundTag":"value-out-of-range","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"}