apache/flink · error · IOException

Could not create RecordReader.

Error message

Could not create RecordReader.

What it means

Wraps an InterruptedException raised while open() creates and initializes the Hadoop RecordReader for a given split (createRecordReader + recordReader.initialize). Flink converts it to an IOException to honour the InputFormat contract. The split has already been assigned to the task when this fires, so it reflects a failure to start reading, usually thread interruption.

Source

Thrown at flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java:195

    public InputSplitAssigner getInputSplitAssigner(HadoopInputSplit[] inputSplits) {
        return new LocatableInputSplitAssigner(inputSplits);
    }

    @Override
    public void open(HadoopInputSplit split) throws IOException {

        // enforce sequential open() calls
        synchronized (OPEN_MUTEX) {
            TaskAttemptContext context =
                    new TaskAttemptContextImpl(configuration, new TaskAttemptID());

            try {
                this.recordReader =
                        this.mapreduceInputFormat.createRecordReader(
                                split.getHadoopInputSplit(), context);
                this.recordReader.initialize(split.getHadoopInputSplit(), context);
            } catch (InterruptedException e) {
                throw new IOException("Could not create RecordReader.", e);
            } finally {
                this.fetched = false;
            }
        }
    }

    @Override
    public boolean reachedEnd() throws IOException {
        if (!this.fetched) {
            fetchNext();
        }
        return !this.hasNext;
    }

    protected void fetchNext() throws IOException {
        try {
            this.hasNext = this.recordReader.nextKeyValue();
        } catch (InterruptedException e) {

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Look at the wrapped InterruptedException in the logs to identify cancellation or upstream task failure that interrupted reader initialization.
2. Verify the input data path is reachable from the TaskManager (HDFS/S3 connectivity, permissions).
3. If using a custom RecordReader, ensure initialize() does not swallow InterruptedException into infinite blocking.
4. Retry/restart the job if the interruption was transient (network blip during open).

try {
    hadoopInputFormat.open(split);
} catch (IOException e) {
    if (e.getCause() instanceof InterruptedException) {
        LOG.warn("RecordReader open interrupted; check task cancellation", e);
        // do not retry blindly — let Flink failover handle it
    } else {
        throw e;
    }
}

When it happens

Trigger: Produced in HadoopInputFormatBase.open(split) when mapreduceInputFormat.createRecordReader(...) or recordReader.initialize(...) throws InterruptedException — typically because the task was cancelled/failing during reader setup, or the reader's initialization performs blocking I/O that was interrupted.

Common situations: Task cancelled while opening a split; reader initialization against a remote filesystem interrupted; restart of a failed task where the previous attempt's interruption is still in flight; custom RecordReader whose initialize blocks and mishandles interrupts.

Related errors

• Could not get KeyValue pair.
• Could not fetch next KeyValue pair.
• Could not get Splits.
• Could not write Record.
• Could not create RecordWriter.

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