{"record":{"id":"1668f9d0cb6b2238","repo":"apache/flink","slug":"could-not-fetch-next-keyvalue-pair","errorCode":null,"errorMessage":"Could not fetch next KeyValue pair.","messagePattern":"Could not fetch next KeyValue pair\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java","lineNumber":214,"sourceCode":"            } finally {\n                this.fetched = false;\n            }\n        }\n    }\n\n    @Override\n    public boolean reachedEnd() throws IOException {\n        if (!this.fetched) {\n            fetchNext();\n        }\n        return !this.hasNext;\n    }\n\n    protected void fetchNext() throws IOException {\n        try {\n            this.hasNext = this.recordReader.nextKeyValue();\n        } catch (InterruptedException e) {\n            throw new IOException(\"Could not fetch next KeyValue pair.\", e);\n        } finally {\n            this.fetched = true;\n        }\n    }\n\n    @Override\n    public void close() throws IOException {\n        if (this.recordReader != null) {\n\n            // enforce sequential close() calls\n            synchronized (CLOSE_MUTEX) {\n                this.recordReader.close();\n            }\n        }\n    }\n\n    // --------------------------------------------------------------------------------------------\n    //  Helper methods","sourceCodeStart":196,"sourceCodeEnd":232,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java#L196-L232","documentation":"Wraps an InterruptedException thrown by recordReader.nextKeyValue() during fetchNext(), the one-record lookahead that backs reachedEnd()/nextRecord(). Re-thrown as an IOException per the InputFormat contract. It signals that record iteration was interrupted, which is almost always a side effect of task lifecycle (cancel/fail) rather than a data defect.","triggerScenarios":"Produced in HadoopInputFormatBase.fetchNext() when this.recordReader.nextKeyValue() throws InterruptedException — e.g. the task was cancelled/failing during record reads, checkpoint barrier handling interrupted the reader, or the reader blocks on I/O and was signalled to stop.","commonSituations":"Job cancellation during steady-state reads; checkpointing under backpressure causing interruption; a slow/blocking RecordReader interrupted by failover; third-party reader that propagates InterruptedException on benign conditions.","solutions":["Read the wrapped InterruptedException in the task logs to find the interrupt source (cancellation, checkpoint failure, failover).","If interruptions align with checkpoints, review checkpoint configuration and whether the reader supports interruptible I/O.","Treat as expected noise if the job was intentionally stopped.","Patch or replace a third-party RecordReader that throws InterruptedException on recoverable conditions."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    while (!hadoopInputFormat.reachedEnd()) {\n        Tuple2<K,V> rec = hadoopInputFormat.nextRecord(reuse);\n        ...\n    }\n} catch (IOException e) {\n    if (e.getCause() instanceof InterruptedException) {\n        // iteration interrupted — usually cancellation or checkpoint; verify task state\n        LOG.warn(\"Record iteration interrupted\", e);\n    } else {\n        throw e;\n    }\n}","preventionTips":["Treat iteration-time InterruptedException-wrapped IOExceptions as lifecycle events.","Check whether the job was cancelled before escalating.","Ensure custom RecordReaders propagate interrupts correctly rather than spinning.","Size checkpoint intervals to avoid frequent mid-record interruptions."],"tags":["hadoop","mapreduce","record-reader","interrupted","iteration"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}