apache/flink · error · IOException

Could not close RecordReader.

Error message

Could not close RecordReader.

What it means

Wraps an InterruptedException raised by recordWriter.close(context) inside close(). NOTE: the message says 'RecordReader' but the code actually closes the RecordWriter — the message is misleading. Flink re-throws as IOException per the OutputFormat contract. After close, the committer commits the task and tmp files are renamed, so an interrupt here can also leave partial tmp output.

Source

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

                throw new IOException("Could not create RecordWriter.", e);
            }
        }
    }

    /**
     * commit the task by moving the output file out from the temporary directory.
     *
     * @throws java.io.IOException
     */
    @Override
    public void close() throws IOException {

        // enforce sequential close() calls
        synchronized (CLOSE_MUTEX) {
            try {
                this.recordWriter.close(this.context);
            } catch (InterruptedException e) {
                throw new IOException("Could not close RecordReader.", e);
            }

            if (this.outputCommitter.needsTaskCommit(this.context)) {
                this.outputCommitter.commitTask(this.context);
            }

            Path outputPath = new Path(this.configuration.get("mapred.output.dir"));

            // rename tmp-file to final name
            FileSystem fs = FileSystem.get(outputPath.toUri(), this.configuration);

            String taskNumberStr = Integer.toString(this.taskNumber);
            String tmpFileTemplate = "tmp-r-00000";
            String tmpFile =
                    tmpFileTemplate.substring(0, 11 - taskNumberStr.length()) + taskNumberStr;

            if (fs.exists(new Path(outputPath.toString() + "/" + tmpFile))) {
                fs.rename(

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Read the wrapped InterruptedException in the logs; cancellation/failover is the usual cause, not a sink bug.
2. Check for leftover tmp files (mapreduce.output.basename 'tmp') under the output dir that close() did not get to rename, and clean them up if the job was not retried.
3. Verify the output filesystem is reachable so close() does not block indefinitely.
4. If the message is confusing your alerting, note that 'Could not close RecordReader.' from this sink always refers to the output RecordWriter.

try {
    hadoopOutputFormat.close();
} catch (IOException e) {
    if (e.getCause() instanceof InterruptedException) {
        // close() of the RecordWriter was interrupted — note tmp files may be left behind
        LOG.warn("Output close interrupted (message says 'RecordReader' but it is the writer);"
            + " check for leftover tmp files in " + outputDir, e);
    } else {
        throw e;
    }
}

When it happens

Trigger: Produced in HadoopOutputFormatBase.close() when this.recordWriter.close(this.context) throws InterruptedException — e.g. the task was cancelled/failing during final flush+close, the underlying filesystem close was interrupted, or the writer's commit path blocked and was interrupted.

Common situations: Job cancellation during sink close; task failover interrupting the final write/flush; slow output filesystem where close() blocks; custom RecordWriter whose close mishandles interrupts. Note the misleading 'RecordReader' wording — this is the output writer closing.

Related errors

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

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