apache/flink · error · IOException

Could not get KeyValue pair.

Error message

Could not get KeyValue pair.

What it means

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.

Source

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

            Class<K> key,
            Class<V> value)
            throws IOException {
        super(mapreduceInputFormat, key, value, Job.getInstance());
    }

    @Override
    public Tuple2<K, V> nextRecord(Tuple2<K, V> record) throws IOException {
        if (!this.fetched) {
            fetchNext();
        }
        if (!this.hasNext) {
            return null;
        }
        try {
            record.f0 = recordReader.getCurrentKey();
            record.f1 = recordReader.getCurrentValue();
        } catch (InterruptedException e) {
            throw new IOException("Could not get KeyValue pair.", e);
        }
        this.fetched = false;

        return record;
    }

    @Override
    public TypeInformation<Tuple2<K, V>> getProducedType() {
        return new TupleTypeInfo<Tuple2<K, V>>(
                TypeExtractor.createTypeInfo(keyClass), TypeExtractor.createTypeInfo(valueClass));
    }
}

View on GitHub (pinned to 2f3c205e92)

Solutions

1. 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.
2. 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.
3. If the job was intentionally cancelled, treat this as expected noise and suppress in alerting.
4. Upgrade or patch a third-party InputFormat whose nextKeyValue/getCurrent* methods mishandle thread interruption.

try {
    Tuple2<K,V> rec = hadoopInputFormat.nextRecord(reuse);
} catch (IOException e) {
    if (e.getCause() instanceof InterruptedException) {
        // input was interrupted — usually job cancel/fail; check task state before treating as fatal
        LOG.warn("Input read interrupted; likely task cancellation", e);
    } else {
        throw e;
    }
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Could not get Splits.
• Could not create RecordReader.
• Could not fetch next KeyValue pair.
• Unable to instantiate the hadoop input format
• Could not write Record.

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