What causes "Interrupted while enqueuing error row for error sink" in apache/seatunnel?

Expected during job cancellation - no user action needed unless the job was not intentionally cancelled

apache/seatunnel · warning · RuntimeException

Interrupted while enqueuing error row for error sink

Error message

Interrupted while enqueuing error row for error sink

What it means

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.

Source

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

                    pendingRows.incrementAndGet();
                    enqueueWithBlockPolicy(ctx, errorRow);
                    break;
                case FAIL:
                default:
                    pendingRows.incrementAndGet();
                    if (!queue.offer(errorRow)) {
                        pendingRows.decrementAndGet();
                        throw new RuntimeException(
                                String.format(
                                        "Error queue overflow for stage [%s], plugin [%s]",
                                        ctx.getStage(), ctx.getPluginName()));
                    }
                    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;

View on GitHub (pinned to cf67b549a7)

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

Defensive patterns

Strategy: try-catch

Try / catch

try {
    writer.write(row);
} catch (RuntimeException e) {
    if (e.getCause() instanceof InterruptedException) {
        Thread.currentThread().interrupt(); // preserve interrupt status upstream
    }
    throw e;
}

Prevention

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

When it happens

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

Common situations: Cancelling a running job while the error sink is backpressured; engine checkpoint timeout interrupting blocked tasks; thread pool shutdown during graceful stop.

Related errors

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