apache/flink · error · IOException

Could not fetch next KeyValue pair.

Error message

Could not fetch next KeyValue pair.

What it means

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.

Source

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

            } finally {
                this.fetched = false;
            }
        }
    }

    @Override
    public boolean reachedEnd() throws IOException {
        if (!this.fetched) {
            fetchNext();
        }
        return !this.hasNext;
    }

    protected void fetchNext() throws IOException {
        try {
            this.hasNext = this.recordReader.nextKeyValue();
        } catch (InterruptedException e) {
            throw new IOException("Could not fetch next KeyValue pair.", e);
        } finally {
            this.fetched = true;
        }
    }

    @Override
    public void close() throws IOException {
        if (this.recordReader != null) {

            // enforce sequential close() calls
            synchronized (CLOSE_MUTEX) {
                this.recordReader.close();
            }
        }
    }

    // --------------------------------------------------------------------------------------------
    //  Helper methods

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Read the wrapped InterruptedException in the task logs to find the interrupt source (cancellation, checkpoint failure, failover).
2. If interruptions align with checkpoints, review checkpoint configuration and whether the reader supports interruptible I/O.
3. Treat as expected noise if the job was intentionally stopped.
4. Patch or replace a third-party RecordReader that throws InterruptedException on recoverable conditions.

try {
    while (!hadoopInputFormat.reachedEnd()) {
        Tuple2<K,V> rec = hadoopInputFormat.nextRecord(reuse);
        ...
    }
} catch (IOException e) {
    if (e.getCause() instanceof InterruptedException) {
        // iteration interrupted — usually cancellation or checkpoint; verify task state
        LOG.warn("Record iteration interrupted", e);
    } else {
        throw e;
    }
}

When it happens

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

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

Related errors

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

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