What causes "Interrupted while waiting for error sink queue" in apache/seatunnel?

Expected during cancellation - no action if the job was intentionally stopped

apache/seatunnel · warning · RuntimeException

Interrupted while waiting for error sink queue

Error message

Interrupted while waiting for error sink queue

What it means

RuntimeException thrown by waitForPendingRows when its Thread.sleep(10ms) poll loop is interrupted while waiting for the error sink queue to drain during flush. The interrupt flag is restored and the exception is rethrown wrapped, aborting the flush operation.

Source

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

    private void waitForPendingRows(long timeoutMillis) throws Exception {
        AtomicInteger currentPendingRows = this.pendingRows;
        if (currentPendingRows == null) {
            return;
        }
        long deadline = System.currentTimeMillis() + timeoutMillis;
        while (currentPendingRows.get() > 0) {
            throwWorkerFailureIfAny();
            if (System.currentTimeMillis() >= deadline) {
                throw new RuntimeException(
                        String.format(
                                "Timed out waiting for error sink queue to drain. jobId=%d, pluginName=%s, pendingRows=%d",
                                jobId, sinkConfig.getPluginName(), currentPendingRows.get()));
            }
            try {
                Thread.sleep(10L);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("Interrupted while waiting for error sink queue", e);
            }
        }
    }

    private void throwWorkerFailureIfAny() throws Exception {
        Throwable failure = workerFailure;
        if (failure == null) {
            return;
        }
        if (failure instanceof Error) {
            throw (Error) failure;
        }
        if (failure instanceof Exception) {
            throw (Exception) failure;
        }
        throw new RuntimeException(failure);
    }

View on GitHub (pinned to cf67b549a7)

Solutions

Expected during cancellation - no action if the job was intentionally stopped
Raise checkpoint timeout if checkpoints are being cancelled during heavy flush
Speed up the error sink or drain it earlier so flush waits are short
Check for external code interrupting task threads unexpectedly

Defensive patterns

Strategy: try-catch

Try / catch

try {
    writer.flush();
} catch (RuntimeException e) {
    if (e.getCause() instanceof InterruptedException) {
        Thread.currentThread().interrupt();
        LOG.warn("Flush interrupted while draining error sink queue");
    }
    throw e;
}

Prevention

Treat interrupts during flush as expected during cancellation
Increase checkpoint timeout to reduce flush interruption
Drain the error queue incrementally instead of one long flush
Avoid interrupting task threads from custom code

When it happens

Trigger: flushInternal -> waitForPendingRows is sleeping between pendingRows polls and the thread is interrupted - normally by task cancellation, checkpoint timeout, or engine shutdown while a flush is in progress.

Common situations: Cancelling a job during checkpoint flush; checkpoint timeout killing a slow flush; engine worker shutdown while error rows are still pending.

Related errors

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