apache/seatunnel · error · RuntimeException

Error sink is closing for stage [%s], plugin [%s]

Error message

Error sink is closing for stage [%s], plugin [%s]

What it means

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.

Source

Thrown at seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/error/DefaultErrorSinkWriter.java:195

                    }
                    break;
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            pendingRows.decrementAndGet();
            throw new RuntimeException("Interrupted while enqueuing error row for error sink", e);
        }
        return true;
    }

    private void enqueueWithBlockPolicy(RowErrorContext ctx, SeaTunnelRow errorRow)
            throws Exception {
        boolean enqueued = false;
        try {
            while (true) {
                throwWorkerFailureIfAny();
                if (closed) {
                    throw new RuntimeException(
                            String.format(
                                    "Error sink is closing for stage [%s], plugin [%s]",
                                    ctx.getStage(), ctx.getPluginName()));
                }
                if (queue.offer(errorRow, BLOCK_OFFER_RETRY_MILLIS, TimeUnit.MILLISECONDS)) {
                    enqueued = true;
                    return;
                }
            }
        } catch (InterruptedException e) {
            throw e;
        } catch (Exception | Error e) {
            if (!enqueued) {
                pendingRows.decrementAndGet();
            }
            throw e;
        }
    }

View on GitHub (pinned to cf67b549a7)

Solutions

1. Ensure all rows are written (and flushed) before triggering close on the task
2. Check for premature close caused by task failure elsewhere in the pipeline
3. Coordinate stop ordering so upstream stops producing error rows before the sink writer closes
4. If seen during normal cancellation, it is benign - rely on job restart from checkpoint

if (errorSinkWriter.isClosed()) { /* do not enqueue; drop or buffer elsewhere */ }

try {
    writer.write(row);
} catch (RuntimeException e) {
    if (e.getMessage() != null && e.getMessage().contains("is closing")) {
        LOG.warn("Write raced with error sink close; row discarded: {}", row);
    }
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Failed to invoke ${clazz.getName()}.${methodName}()
• The split fetcher manager has closed.
• close event processor error
• schema-change-after checkpoint is already completed, job id:
• Job is trying to leave terminal state ${current}

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/85f74a0940ec5767. Report an issue: GitHub.