apache/seatunnel · error · RuntimeException

Timed out waiting for error sink queue to drain. jobId=%d, p

Error message

Timed out waiting for error sink queue to drain. jobId=%d, pluginName=%s, pendingRows=%d

What it means

RuntimeException thrown by waitForPendingRows (called from flushInternal) when the error sink queue still has pending rows after the configured drain timeout. It polls pendingRows every 10ms up to a deadline, rechecking worker failure; on timeout it reports jobId, plugin and remaining pending row count, indicating the error sink could not keep up or the worker is stuck.

Source

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

                workerThread.join(Math.min(5_000L, timeoutMillis));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.warn(
                        "Interrupted while waiting for error sink worker to close after interrupt");
            }
        }
    }

    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;
        }

View on GitHub (pinned to cf67b549a7)

Solutions

Increase the error-sink flush/drain timeout configuration
Reduce dirty-row volume or increase error sink parallelism/throughput
Investigate the error sink's downstream latency (network, target DB)
Enable error-sink metrics/logging to confirm the worker is progressing, not hung

Example fix

# before
error-sink {
  ErrorQueueCapacity = 100
  flush_interval = 1000
}
# after
error-sink {
  ErrorQueueCapacity = 10000
  flush_interval = 10000
  drain_timeout = 60000
}

Defensive patterns

Strategy: retry

Validate before calling

if (pendingRows.get() > errorQueueCapacity * 0.9) {
    LOG.warn("Error sink near capacity; consider flushing early");
}

Try / catch

try {
    writer.flush();
} catch (RuntimeException e) {
    if (e.getMessage() != null && e.getMessage().contains("Timed out waiting")) {
        LOG.warn("Error sink drain timeout; retrying after worker recovery");
        // retry flush or fail checkpoint to trigger restart
    }
    throw e;
}

Prevention

Tune drain timeout to exceed worst-case error sink latency
Prevent huge dirty-row bursts immediately before checkpoints
Monitor error sink throughput and lag
Use a faster local error sink if the remote one is slow

When it happens

Trigger: flush() is invoked (checkpoint or close) and pendingRows.get() > 0 when System.currentTimeMillis() passes the deadline; each poll first rethrows any worker failure, so this only fires when the worker is alive but too slow or blocked.

Common situations: Huge dirty-row burst just before checkpoint; error sink writing to a slow/hung remote system; drain timeout configured too small for the sink throughput.

Understand the failure class

Background: Request timed out: what client-side request timeouts mean across libraries (Request timed out, TIMED_OUT, APITimeoutError) — this error's family across 39 libraries.

Timeouts: ETIMEDOUT, deadlines, and hung requests — what actually expires when a request times out.

Related errors

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