{"record":{"id":"16ddc1bf89372564","repo":"apache/flink","slug":"could-not-get-keyvalue-pair","errorCode":null,"errorMessage":"Could not get KeyValue pair.","messagePattern":"Could not get 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/HadoopInputFormat.java","lineNumber":72,"sourceCode":"            Class<K> key,\n            Class<V> value)\n            throws IOException {\n        super(mapreduceInputFormat, key, value, Job.getInstance());\n    }\n\n    @Override\n    public Tuple2<K, V> nextRecord(Tuple2<K, V> record) throws IOException {\n        if (!this.fetched) {\n            fetchNext();\n        }\n        if (!this.hasNext) {\n            return null;\n        }\n        try {\n            record.f0 = recordReader.getCurrentKey();\n            record.f1 = recordReader.getCurrentValue();\n        } catch (InterruptedException e) {\n            throw new IOException(\"Could not get KeyValue pair.\", e);\n        }\n        this.fetched = false;\n\n        return record;\n    }\n\n    @Override\n    public TypeInformation<Tuple2<K, V>> getProducedType() {\n        return new TupleTypeInfo<Tuple2<K, V>>(\n                TypeExtractor.createTypeInfo(keyClass), TypeExtractor.createTypeInfo(valueClass));\n    }\n}\n","sourceCodeStart":54,"sourceCodeEnd":85,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java#L54-L85","documentation":"Wraps an InterruptedException raised by Hadoop's RecordReader.getCurrentKey()/getCurrentValue() while nextRecord() reads a key/value pair from the mapreduce InputFormat. Flink converts the checked InterruptedException into an IOException so it propagates through the InputFormat contract. The underlying cause is almost always the reader thread being interrupted, not a data problem.","triggerScenarios":"Produced when HadoopInputFormat.nextRecord() calls recordReader.getCurrentKey() or recordReader.getCurrentValue() and the Hadoop RecordReader throws InterruptedException — typically because the task was cancelled/failing, a checkpoint barrier interrupted processing, or the reader's internal threading was signalled to stop.","commonSituations":"Job cancellation mid-read; task failure that triggers interruption of the input thread; a source reader that blocks and gets interrupted on timeout; stress under backpressure where a task is killed and its threads interrupted.","solutions":["Check the taskmanager logs for the root cause (cancellation, checkpoint failure, OOM) recorded alongside the wrapped InterruptedException; the input format itself is rarely at fault.","If interruptions recur during checkpoints, review checkpoint interval and the Hadoop RecordReader's interruptibility, and ensure the reader does not hold blocking I/O across barrier alignment.","If the job was intentionally cancelled, treat this as expected noise and suppress in alerting.","Upgrade or patch a third-party InputFormat whose nextKeyValue/getCurrent* methods mishandle thread interruption."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    Tuple2<K,V> rec = hadoopInputFormat.nextRecord(reuse);\n} catch (IOException e) {\n    if (e.getCause() instanceof InterruptedException) {\n        // input was interrupted — usually job cancel/fail; check task state before treating as fatal\n        LOG.warn(\"Input read interrupted; likely task cancellation\", e);\n    } else {\n        throw e;\n    }\n}","preventionTips":["Treat InterruptedException-wrapped IOExceptions as lifecycle noise unless they recur.","Correlate the exception with task cancellation/failover events in the logs.","Avoid blocking, non-interruptible I/O in custom RecordReaders.","Set checkpoint intervals so readers are not interrupted mid-record under backpressure."],"tags":["hadoop","mapreduce","input-format","interrupted","record-reader"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}