apache/flink · error · RuntimeException

Failed to close current reader

Error message

Failed to close current reader

What it means

Thrown as a RuntimeException by HybridSourceReader.setCurrentReader when closing the current underlying SourceReader (before switching to the next source in the hybrid pipeline) throws an exception. setCurrentReader calls currentReader.close() when transitioning from one source to the next; any failure during the underlying reader's cleanup is wrapped and rethrown to prevent continuing with a half-closed reader.

Source

Thrown at flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceReader.java:212

    @Override
    public void close() throws Exception {
        if (currentReader != null) {
            currentReader.close();
        }
        LOG.debug(
                "Reader closed: subtask={} sourceIndex={} currentReader={}",
                readerContext.getIndexOfSubtask(),
                currentSourceIndex,
                currentReader);
    }

    private void setCurrentReader(int index) {
        Preconditions.checkArgument(index != currentSourceIndex);
        if (currentReader != null) {
            try {
                currentReader.close();
            } catch (Exception e) {
                throw new RuntimeException("Failed to close current reader", e);
            }
            LOG.debug(
                    "Reader closed: subtask={} sourceIndex={} currentReader={}",
                    readerContext.getIndexOfSubtask(),
                    currentSourceIndex,
                    currentReader);
        }
        // TODO: track previous readers splits till checkpoint
        Source source = switchedSources.sourceOf(index);
        SourceReader<T, ?> reader;
        try {
            reader = source.createReader(readerContext);
        } catch (Exception e) {
            throw new RuntimeException("Failed to create reader", e);
        }
        // currentReader must be switched before `addSplits` is called.
        currentSourceIndex = index;
        currentReader = reader;

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Check the wrapped exception cause to identify which underlying source reader's close() failed.
2. If the underlying source has a known close() bug, update to a patched version of that connector.
3. For custom source implementations, ensure SourceReader.close() catches and logs internal errors rather than propagating them.
4. If the error is transient (network hiccup during transition), the job will failover and retry from the last checkpoint.

// This error occurs internally in HybridSourceReader; handle at job level
try {
    env.execute("hybridSourceJob");
} catch (Exception e) {
    Throwable cause = ExceptionUtils.findThrowable(e, RuntimeException.class).orElse(e);
    if (cause.getMessage() != null && cause.getMessage().equals("Failed to close current reader")) {
        // underlying source reader close failed — job will failover and retry
        log.error("Source transition failed during reader close: {}", cause.getCause());
    }
}

When it happens

Trigger: HybridSource transitions from source N to source N+1, and the SourceReader.close() of source N throws — e.g., the reader's network connection cleanup fails, or its background threads do not shut down cleanly.

Common situations: Underlying source reader has a resource leak or cleanup bug; network/filesystem connection is lost during source transition; the reader's close() is not idempotent and throws on double-close; a custom source implementation has improper exception handling in close().

Related errors

• Failed to create reader
• Could not close the JAR file: {t.getMessage()}
• The bytes are serialized with version %d, while this deseria
• Failed to create enumerator for sourceIndex={currentSourceIn
• Invalid version %d

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/9ddc65ae2b7f54b6. Report an issue: GitHub.