{"record":{"id":"922fed69e3f005f5","repo":"apache/seatunnel","slug":"interrupted-while-enqueuing-error-row-for-error-si","errorCode":null,"errorMessage":"Interrupted while enqueuing error row for error sink","messagePattern":"Interrupted while enqueuing error row for error sink","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":183,"sourceCode":"                    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 {\n            while (true) {\n                throwWorkerFailureIfAny();\n                if (closed) {\n                    throw new RuntimeException(\n                            String.format(\n                                    \"Error sink is closing for stage [%s], plugin [%s]\",\n                                    ctx.getStage(), ctx.getPluginName()));\n                }\n                if (queue.offer(errorRow, BLOCK_OFFER_RETRY_MILLIS, TimeUnit.MILLISECONDS)) {\n                    enqueued = true;","sourceCodeStart":165,"sourceCodeEnd":201,"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#L165-L201","documentation":"Thrown when the thread submitting an error row is interrupted while waiting to enqueue into the error sink queue (blocking path inside writeAndCheckAccepted). The method restores the interrupt flag, rolls back the pendingRows counter, and rethrows as RuntimeException so the job framework knows the write did not complete.","triggerScenarios":"write() -> writeAndCheckAccepted() is blocked enqueuing an error row (blocking overflow policy) and Thread.interrupt() arrives - typically from task cancellation, job stopping, or checkpoint-timeout cancellation by the Zeta engine.","commonSituations":"Cancelling a running job while the error sink is backpressured; engine checkpoint timeout interrupting blocked tasks; thread pool shutdown during graceful stop.","solutions":["Expected during job cancellation - no user action needed unless the job was not intentionally cancelled","Check checkpoint timeout settings if checkpoint cancellation triggers it frequently","Reduce error sink backpressure (bigger queue, faster sink) so writes rarely block","Ensure the error-sink thread is not permanently blocked on a hung downstream"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    writer.write(row);\n} catch (RuntimeException e) {\n    if (e.getCause() instanceof InterruptedException) {\n        Thread.currentThread().interrupt(); // preserve interrupt status upstream\n    }\n    throw e;\n}","preventionTips":["Expect interrupts during job cancellation - treat as normal","Set generous checkpoint timeouts to reduce mid-write interrupts","Avoid custom code that interrupts task threads","Reduce error sink blocking so writes complete quickly"],"tags":["error-sink","interrupted-exception","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"}