apache/flink · warning · RuntimeException

Interrupted while waiting for the previous batch to be consu

Error message

Interrupted while waiting for the previous batch to be consumed

What it means

In AbstractAvroBulkFormat's batch reader, readBatch() takes a datum/mutator instance from a bounded pool via pool.pollEntry(), which blocks while all instances are consumed by in-flight batches. If the reader thread is interrupted while waiting, the interrupt flag is re-set and a RuntimeException wraps the failure — normally a job cancellation side effect, not a data problem.

Source

Thrown at flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AbstractAvroBulkFormat.java:150

        private DataFileReader<A> createReaderFromPath(Path path) throws IOException {
            FileSystem fileSystem = path.getFileSystem();
            DatumReader<A> datumReader = new GenericDatumReader<>(null, readerSchema);
            SeekableInput in =
                    new FSDataInputStreamWrapper(
                            fileSystem.open(path), fileSystem.getFileStatus(path).getLen());
            return (DataFileReader<A>) DataFileReader.openReader(in, datumReader);
        }

        @Nullable
        @Override
        public RecordIterator<T> readBatch() throws IOException {
            A reuse;
            try {
                reuse = pool.pollEntry();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(
                        "Interrupted while waiting for the previous batch to be consumed", e);
            }

            if (!readNextBlock()) {
                pool.recycler().recycle(reuse);
                return null;
            }

            currentBlockStart = reader.previousSync();
            Iterator<T> iterator =
                    new AvroBlockIterator(
                            reader.getBlockCount() - currentRecordsToSkip,
                            reader,
                            reuse,
                            converter);
            long recordsToSkip = currentRecordsToSkip;
            currentRecordsToSkip = 0;
            return new IteratorResultIterator<>(

View on GitHub (pinned to 2f3c205e92)

Solutions

1. If it appears during cancellation/shutdown, it is benign — ensure shutdown code does not rethrow it as a job failure.
2. Drain or release record iterators promptly (call releaseBatch/close on iterators) so the pool is not starved.
3. Increase the reader pool size if legitimate waits are long (config depending on format setup).
4. For frameworks: catch RuntimeException with InterruptedException cause in source loops and exit cleanly when Thread.interrupted().

Example fix

// before
catch (RuntimeException e) { throw e; } // turns cancellation into failure

// after
catch (RuntimeException e) {
    if (e.getCause() instanceof InterruptedException) {
        Thread.currentThread().interrupt();
        return; // cooperative shutdown
    }
    throw e;
}

try {
    batch = reader.readBatch();
} catch (RuntimeException e) {
    if (e.getCause() instanceof InterruptedException) {
        Thread.currentThread().interrupt();
        return; // cooperative cancel
    }
    throw e;
}

When it happens

Trigger: Reading an Avro file with this bulk format while record iterators from previous batches have not been fully drained (pool empty), and the task gets interrupted — job cancel, failover restart, or source shutdown. pollEntry() throws InterruptedException and is converted here.

Common situations: Job cancellation while reading a large Avro split; bounded object pool sized too small relative to concurrent batches, making waits routine and thus likely to overlap with cancellation; failover restarting source tasks mid-read.

Related errors

• Interrupted while uploading object for key: {}
• interrupted while acquiring lock
• Interrupted
• Error while getting the file registered under '${name}' from
• Interrupted when untarring file {inFilePath}

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