{"record":{"id":"85f74a0940ec5767","repo":"apache/seatunnel","slug":"error-sink-is-closing-for-stage-s-plugin-s","errorCode":null,"errorMessage":"Error sink is closing for stage [%s], plugin [%s]","messagePattern":"Error sink is closing 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":195,"sourceCode":"                    }\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;\n                    return;\n                }\n            }\n        } catch (InterruptedException e) {\n            throw e;\n        } catch (Exception | Error e) {\n            if (!enqueued) {\n                pendingRows.decrementAndGet();\n            }\n            throw e;\n        }\n    }","sourceCodeStart":177,"sourceCodeEnd":213,"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#L177-L213","documentation":"enqueueWithBlockPolicy loops offering rows to the error queue with a retry timeout; before each retry it rechecks worker failure and the closed flag. If the error sink writer is closing (closed == true) while a row is still being enqueued, it throws this RuntimeException naming the stage and plugin - a clean-shutdown guard preventing rows from being silently dropped into a closing sink.","triggerScenarios":"A row is enqueued with the blocking policy while close()/shutdown of DefaultErrorSinkWriter is in progress: throwWorkerFailureIfAny passes, but closed has been set, so the loop throws instead of waiting.","commonSituations":"Writing error rows concurrently with job stop/complete; task finish signal arriving while a final batch of dirty rows is still being routed; races between flush/close and late error rows.","solutions":["Ensure all rows are written (and flushed) before triggering close on the task","Check for premature close caused by task failure elsewhere in the pipeline","Coordinate stop ordering so upstream stops producing error rows before the sink writer closes","If seen during normal cancellation, it is benign - rely on job restart from checkpoint"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"if (errorSinkWriter.isClosed()) { /* do not enqueue; drop or buffer elsewhere */ }","typeGuard":null,"tryCatchPattern":"try {\n    writer.write(row);\n} catch (RuntimeException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"is closing\")) {\n        LOG.warn(\"Write raced with error sink close; row discarded: {}\", row);\n    }\n}","preventionTips":["Flush all rows before closing the task","Stop producing error rows before triggering sink close","Avoid writing from threads other than the task thread during shutdown","Rely on checkpoint recovery if rows are lost during stop"],"tags":["error-sink","race-condition","shutdown","zeta-engine"],"backgroundTag":"invalid-state-transition","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"}