apache/flink · error · RuntimeException

Failed to create reader

Error message

Failed to create reader

What it means

Thrown as a RuntimeException by HybridSourceReader.setCurrentReader when calling source.createReader(readerContext) for the next source in the hybrid pipeline throws an exception. After closing the previous reader, setCurrentReader instantiates the new SourceReader; any failure during reader creation (misconfiguration, missing resources, initialization error) is wrapped and rethrown.

Source

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

        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;
        // add restored splits
        if (!restoredSplits.isEmpty()) {
            List<HybridSourceSplit> splits = new ArrayList<>(restoredSplits.size());
            Iterator<HybridSourceSplit> it = restoredSplits.iterator();
            while (it.hasNext()) {
                HybridSourceSplit hybridSplit = it.next();
                if (hybridSplit.sourceIndex() == index) {
                    splits.add(hybridSplit);
                    it.remove();
                }
            }
            addSplits(splits);
        }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Check the wrapped exception cause to identify what failed in createReader — it is typically the source-specific initialization error.
2. Verify the configuration for the failing source in the HybridSource chain (connection strings, credentials, etc.).
3. Ensure switchedSources is correctly populated with all sources in the pipeline via HybridSourceSource.switchedSources.
4. If the error is environmental (e.g., external system temporarily unavailable), the job failover mechanism will retry.

// Validate source configuration before building HybridSource
Source<?, ?, ?> nextSource = sources.get(nextIndex);
if (nextSource == null) {
    throw new IllegalArgumentException("No source configured at index " + nextIndex);
}
// Test reader creation in a try-with-resources
try (SourceReader<?, ?> testReader = nextSource.createReader(testContext)) {
    // reader created successfully
}

// 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 create reader")) {
        // check the wrapped cause for the source-specific initialization error
        log.error("Reader creation failed for source in chain: {}", cause.getCause());
    }
}

When it happens

Trigger: HybridSource transitions to source index N, and switchedSources.sourceOf(N).createReader(context) throws — e.g., the source factory cannot connect to the external system, required configuration is missing, or the source implementation has a bug in its constructor.

Common situations: The next source in the HybridSource chain has incorrect connection configuration (e.g., wrong Kafka brokers, missing credentials); the source's createReader requires resources not available at transition time; switchedSources does not contain the expected source at the given index; custom source init code throws.

Related errors

• Failed to close current reader
• The program's entry point class '{className}' threw an error
• The program's entry point class '{className}' caused an exce
• The bytes are serialized with version %d, while this deseria
• Failed to create enumerator for sourceIndex={currentSourceIn

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