{"record":{"id":"7c7566ca3f650642","repo":"apache/flink","slug":"could-not-close-recordreader","errorCode":null,"errorMessage":"Could not close RecordReader.","messagePattern":"Could not close RecordReader\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java","lineNumber":182,"sourceCode":"                throw new IOException(\"Could not create RecordWriter.\", e);\n            }\n        }\n    }\n\n    /**\n     * commit the task by moving the output file out from the temporary directory.\n     *\n     * @throws java.io.IOException\n     */\n    @Override\n    public void close() throws IOException {\n\n        // enforce sequential close() calls\n        synchronized (CLOSE_MUTEX) {\n            try {\n                this.recordWriter.close(this.context);\n            } catch (InterruptedException e) {\n                throw new IOException(\"Could not close RecordReader.\", e);\n            }\n\n            if (this.outputCommitter.needsTaskCommit(this.context)) {\n                this.outputCommitter.commitTask(this.context);\n            }\n\n            Path outputPath = new Path(this.configuration.get(\"mapred.output.dir\"));\n\n            // rename tmp-file to final name\n            FileSystem fs = FileSystem.get(outputPath.toUri(), this.configuration);\n\n            String taskNumberStr = Integer.toString(this.taskNumber);\n            String tmpFileTemplate = \"tmp-r-00000\";\n            String tmpFile =\n                    tmpFileTemplate.substring(0, 11 - taskNumberStr.length()) + taskNumberStr;\n\n            if (fs.exists(new Path(outputPath.toString() + \"/\" + tmpFile))) {\n                fs.rename(","sourceCodeStart":164,"sourceCodeEnd":200,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java#L164-L200","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Read the wrapped InterruptedException in the logs; cancellation/failover is the usual cause, not a sink bug.","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.","Verify the output filesystem is reachable so close() does not block indefinitely.","If the message is confusing your alerting, note that 'Could not close RecordReader.' from this sink always refers to the output RecordWriter."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    hadoopOutputFormat.close();\n} catch (IOException e) {\n    if (e.getCause() instanceof InterruptedException) {\n        // close() of the RecordWriter was interrupted — note tmp files may be left behind\n        LOG.warn(\"Output close interrupted (message says 'RecordReader' but it is the writer);\"\n            + \" check for leftover tmp files in \" + outputDir, e);\n    } else {\n        throw e;\n    }\n}","preventionTips":["Remember the message 'Could not close RecordReader.' from this sink actually refers to the output RecordWriter.","After an interrupted close, check for un-renamed tmp files (mapreduce.output.basename 'tmp').","Verify output filesystem reachability so close() does not block.","Treat close-time interruptions as task-lifecycle events."],"tags":["hadoop","mapreduce","output-format","interrupted","record-writer"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}