{"record":{"id":"36255616ac125f16","repo":"apache/flink","slug":"interrupted-while-waiting-for-the-previous-batch-t","errorCode":null,"errorMessage":"Interrupted while waiting for the previous batch to be consumed","messagePattern":"Interrupted while waiting for the previous batch to be consumed","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"warning","filePath":"flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AbstractAvroBulkFormat.java","lineNumber":150,"sourceCode":"\n        private DataFileReader<A> createReaderFromPath(Path path) throws IOException {\n            FileSystem fileSystem = path.getFileSystem();\n            DatumReader<A> datumReader = new GenericDatumReader<>(null, readerSchema);\n            SeekableInput in =\n                    new FSDataInputStreamWrapper(\n                            fileSystem.open(path), fileSystem.getFileStatus(path).getLen());\n            return (DataFileReader<A>) DataFileReader.openReader(in, datumReader);\n        }\n\n        @Nullable\n        @Override\n        public RecordIterator<T> readBatch() throws IOException {\n            A reuse;\n            try {\n                reuse = pool.pollEntry();\n            } catch (InterruptedException e) {\n                Thread.currentThread().interrupt();\n                throw new RuntimeException(\n                        \"Interrupted while waiting for the previous batch to be consumed\", e);\n            }\n\n            if (!readNextBlock()) {\n                pool.recycler().recycle(reuse);\n                return null;\n            }\n\n            currentBlockStart = reader.previousSync();\n            Iterator<T> iterator =\n                    new AvroBlockIterator(\n                            reader.getBlockCount() - currentRecordsToSkip,\n                            reader,\n                            reuse,\n                            converter);\n            long recordsToSkip = currentRecordsToSkip;\n            currentRecordsToSkip = 0;\n            return new IteratorResultIterator<>(","sourceCodeStart":132,"sourceCodeEnd":168,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AbstractAvroBulkFormat.java#L132-L168","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["If it appears during cancellation/shutdown, it is benign — ensure shutdown code does not rethrow it as a job failure.","Drain or release record iterators promptly (call releaseBatch/close on iterators) so the pool is not starved.","Increase the reader pool size if legitimate waits are long (config depending on format setup).","For frameworks: catch RuntimeException with InterruptedException cause in source loops and exit cleanly when Thread.interrupted()."],"exampleFix":"// before\ncatch (RuntimeException e) { throw e; } // turns cancellation into failure\n\n// after\ncatch (RuntimeException e) {\n    if (e.getCause() instanceof InterruptedException) {\n        Thread.currentThread().interrupt();\n        return; // cooperative shutdown\n    }\n    throw e;\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    batch = reader.readBatch();\n} catch (RuntimeException e) {\n    if (e.getCause() instanceof InterruptedException) {\n        Thread.currentThread().interrupt();\n        return; // cooperative cancel\n    }\n    throw e;\n}","preventionTips":["Release record iterators promptly so the pool never starves.","Treat this during cancellation as expected shutdown noise.","Size the datum pool to your concurrent-batch count."],"tags":["avro","format","interruption","cancellation","concurrency"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}