apache/seatunnel · error · IllegalStateException

ErrorSinkRowWriter is already closed

Error message

ErrorSinkRowWriter is already closed

What it means

SynchronizedErrorSinkRowWriter.write delegates error-row writes to an underlying ErrorSinkRowWriter under a lock. If the writer has already been closed (typically after checkpoint teardown or sink close), any further write throws IllegalStateException 'ErrorSinkRowWriter is already closed'. This protects the closed resource from late writes.

Source

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

/** Thread-safe wrapper for ErrorSinkRowWriter to serialize write/close calls. */
public final class SynchronizedErrorSinkRowWriter<T> implements ErrorSinkRowWriter<T> {

    private static final long serialVersionUID = 1L;

    private final ErrorSinkRowWriter<T> delegate;
    private final Object lock = new Object();
    private volatile boolean closed;

    public SynchronizedErrorSinkRowWriter(ErrorSinkRowWriter<T> delegate) {
        this.delegate = Objects.requireNonNull(delegate, "delegate must not be null");
    }

    @Override
    public void write(RowErrorContext ctx, T row, Throwable t) throws Exception {
        synchronized (lock) {
            if (closed) {
                throw new IllegalStateException("ErrorSinkRowWriter is already closed");
            }
            delegate.write(ctx, row, t);
        }
    }

    @Override
    public boolean writeAndCheckAccepted(RowErrorContext ctx, T row, Throwable t) throws Exception {
        synchronized (lock) {
            if (closed) {
                throw new IllegalStateException("ErrorSinkRowWriter is already closed");
            }
            return delegate.writeAndCheckAccepted(ctx, row, t);
        }
    }

    @Override
    public void flush() throws Exception {
        synchronized (lock) {

View on GitHub (pinned to cf67b549a7)

Solutions

1. Stop feeding error rows before closing the writer; close only after all error paths have finished.
2. Check the closed state (or catch IllegalStateException) in late-emit paths and drop/queue the row instead.
3. Ensure task lifecycle ordering: close error consumers/writers as the last step of task teardown.

Example fix

// before
writer.close();
writer.write(ctx, row, t); // throws
// after
writer.write(ctx, row, t);
writer.close();

null

null

try {
    writer.write(ctx, row, t);
} catch (IllegalStateException e) {
    if (e.getMessage().contains("already closed")) {
        log.warn("Dropping error row after writer close");
    } else throw e;
}

When it happens

Trigger: Calling write(ctx, row, t) after close() was invoked on the writer, e.g. an upstream error path still emitting rows while the task is shutting down.

Common situations: Late error events arriving during task cancellation or checkpoint completion; asynchronous error handlers not stopped before the writer is closed.

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

• Handler not reset
• Python source reader has already been closed
• DataTypeChanger not reset
• MDCContext is already activated
• MDCContext is not activated

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