<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;
    }
@Test
    public void testUnaligned() {
        long firstMessageId = 5;
        long startingOffset = 10L;
        int numRecords = 10;
        ShareFetchResponseData.PartitionData partitionData = new ShareFetchResponseData.PartitionData()
                .setRecords(newRecords(startingOffset, numRecords + 500, firstMessageId))
                .setAcquiredRecords(acquiredRecords(startingOffset + 500, numRecords));

        Deserializers<String, String> deserializers = newStringDeserializers();

        ShareCompletedFetch completedFetch = newShareCompletedFetch(partitionData);

        ShareInFlightBatch<String, String> batch = completedFetch.fetchRecords(deserializers, 10, true);
        List<ConsumerRecord<String, String>> records = batch.getInFlightRecords();
        assertEquals(10, records.size());
        ConsumerRecord<String, String> record = records.get(0);
        assertEquals(510L, record.offset());
        assertEquals(Optional.of((short) 1), record.deliveryCount());
        Acknowledgements acknowledgements = batch.getAcknowledgements();
        assertEquals(0, acknowledgements.size());

        batch = completedFetch.fetchRecords(deserializers, 10, true);
        records = batch.getInFlightRecords();
        assertEquals(0, records.size());
        acknowledgements = batch.getAcknowledgements();
        assertEquals(0, acknowledgements.size());
    }