<K, V> ShareInFlightBatch<K, V> fetchRecords(final Deserializers<K, V> deserializers,
                                                 final int maxRecords,
                                                 final boolean checkCrcs) {
        // Creating an empty ShareInFlightBatch
        ShareInFlightBatch<K, V> inFlightBatch = new ShareInFlightBatch<>(partition);

        if (cachedBatchException != null) {
            // If the event that a CRC check fails, reject the entire record batch because it is corrupt.
            rejectRecordBatch(inFlightBatch, currentBatch);
            inFlightBatch.setException(cachedBatchException);
            cachedBatchException = null;
            return inFlightBatch;
        }

        if (cachedRecordException != null) {
            inFlightBatch.addAcknowledgement(lastRecord.offset(), AcknowledgeType.RELEASE);
            inFlightBatch.setException(cachedRecordException);
            cachedRecordException = null;
            return inFlightBatch;
        }

        if (isConsumed)
            return inFlightBatch;

        initializeNextAcquired();

        try {
            int recordsInBatch = 0;
            while (recordsInBatch < maxRecords) {
                lastRecord = nextFetchedRecord(checkCrcs);
                if (lastRecord == null) {
                    // Any remaining acquired records are gaps
                    while (nextAcquired != null) {
                        inFlightBatch.addGap(nextAcquired.offset);
                        nextAcquired = nextAcquiredRecord();
                    }
                    break;
                }

                while (nextAcquired != null) {
                    if (lastRecord.offset() == nextAcquired.offset) {
                        // It's acquired, so we parse it and add it to the batch
                        Optional<Integer> leaderEpoch = maybeLeaderEpoch(currentBatch.partitionLeaderEpoch());
                        TimestampType timestampType = currentBatch.timestampType();
                        ConsumerRecord<K, V> record = parseRecord(deserializers, partition, leaderEpoch,
                                timestampType, lastRecord, nextAcquired.deliveryCount);
                        inFlightBatch.addRecord(record);
                        recordsRead++;
                        bytesRead += lastRecord.sizeInBytes();
                        recordsInBatch++;

                        nextAcquired = nextAcquiredRecord();
                        break;
                    } else if (lastRecord.offset() < nextAcquired.offset) {
                        // It's not acquired, so we skip it
                        break;
                    } else {
                        // It's acquired, but there's no non-control record at this offset, so it's a gap
                        inFlightBatch.addGap(nextAcquired.offset);
                    }

                    nextAcquired = nextAcquiredRecord();
                }
            }
        } catch (SerializationException se) {
            nextAcquired = nextAcquiredRecord();
            if (inFlightBatch.isEmpty()) {
                inFlightBatch.addAcknowledgement(lastRecord.offset(), AcknowledgeType.RELEASE);
                inFlightBatch.setException(se);
            } else {
                cachedRecordException = se;
                inFlightBatch.setHasCachedException(true);
            }
        } catch (CorruptRecordException e) {
            if (inFlightBatch.isEmpty()) {
                // If the event that a CRC check fails, reject the entire record batch because it is corrupt.
                rejectRecordBatch(inFlightBatch, currentBatch);
                inFlightBatch.setException(e);
            } else {
                cachedBatchException = e;
                inFlightBatch.setHasCachedException(true);
            }
        }

        return inFlightBatch;
    }
The {@link RecordBatch batch} of {@link Record records} is converted to a {@link List list} of
{@link ConsumerRecord consumer records} and returned. {@link BufferSupplier Decompression} and
{@link Deserializer deserialization} of the {@link Record record's} key and value are performed in
this step.

@param deserializers {@link Deserializer}s to use to convert the raw bytes to the expected key and value types
@param maxRecords The number of records to return; the number returned may be {@code 0 <= maxRecords}
@param checkCrcs Whether to check the CRC of fetched records

@return {@link ShareInFlightBatch The ShareInFlightBatch containing records and their acknowledgments}
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java