<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 testCorruptedMessage() {
        // Create one good record and then two "corrupted" records and then another good record.
        try (final MemoryRecordsBuilder builder = MemoryRecords.builder(ByteBuffer.allocate(1024),
                Compression.NONE,
                TimestampType.CREATE_TIME,
                0);
             final UUIDSerializer serializer = new UUIDSerializer()) {
            builder.append(new SimpleRecord(serializer.serialize(TOPIC_NAME, UUID.randomUUID())));
            builder.append(0L, "key".getBytes(), "value".getBytes());
            Headers headers = new RecordHeaders();
            headers.add("hkey", "hvalue".getBytes());
            builder.append(10L, serializer.serialize("key", UUID.randomUUID()), "otherValue".getBytes(), headers.toArray());
            builder.append(new SimpleRecord(serializer.serialize(TOPIC_NAME, UUID.randomUUID())));
            Records records = builder.build();

            ShareFetchResponseData.PartitionData partitionData = new ShareFetchResponseData.PartitionData()
                    .setPartitionIndex(0)
                    .setRecords(records)
                    .setAcquiredRecords(acquiredRecords(0L, 4));

            try (final Deserializers<UUID, UUID> deserializers = newUuidDeserializers()) {
                ShareCompletedFetch completedFetch = newShareCompletedFetch(partitionData);

                // Record 0 is returned by itself because record 1 fails to deserialize
                ShareInFlightBatch<UUID, UUID> batch = completedFetch.fetchRecords(deserializers, 10, false);
                assertNull(batch.getException());
                List<ConsumerRecord<UUID, UUID>> fetchedRecords = batch.getInFlightRecords();
                assertEquals(1, fetchedRecords.size());
                assertEquals(0L, fetchedRecords.get(0).offset());
                Acknowledgements acknowledgements = batch.getAcknowledgements();
                assertEquals(0, acknowledgements.size());

                // Record 1 then results in an empty batch
                batch = completedFetch.fetchRecords(deserializers, 10, false);
                assertEquals(RecordDeserializationException.class, batch.getException().getClass());
                RecordDeserializationException thrown = (RecordDeserializationException) batch.getException();
                assertEquals(RecordDeserializationException.DeserializationExceptionOrigin.KEY, thrown.origin());
                assertEquals(1, thrown.offset());
                assertEquals(TOPIC_NAME, thrown.topicPartition().topic());
                assertEquals(0, thrown.topicPartition().partition());
                assertEquals(0, thrown.timestamp());
                assertArrayEquals("key".getBytes(), org.apache.kafka.common.utils.Utils.toNullableArray(thrown.keyBuffer()));
                assertArrayEquals("value".getBytes(), Utils.toNullableArray(thrown.valueBuffer()));
                assertEquals(0, thrown.headers().toArray().length);
                fetchedRecords = batch.getInFlightRecords();
                assertEquals(0, fetchedRecords.size());
                acknowledgements = batch.getAcknowledgements();
                assertEquals(1, acknowledgements.size());
                assertEquals(AcknowledgeType.RELEASE, acknowledgements.get(1L));

                // Record 2 then results in an empty batch, because record 1 has now been skipped
                batch = completedFetch.fetchRecords(deserializers, 10, false);
                assertEquals(RecordDeserializationException.class, batch.getException().getClass());
                thrown = (RecordDeserializationException) batch.getException();
                assertEquals(RecordDeserializationException.DeserializationExceptionOrigin.VALUE, thrown.origin());
                assertEquals(2L, thrown.offset());
                assertEquals(TOPIC_NAME, thrown.topicPartition().topic());
                assertEquals(0, thrown.topicPartition().partition());
                assertEquals(10L, thrown.timestamp());
                assertNotNull(thrown.keyBuffer());
                assertArrayEquals("otherValue".getBytes(), Utils.toNullableArray(thrown.valueBuffer()));
                fetchedRecords = batch.getInFlightRecords();
                assertEquals(0, fetchedRecords.size());
                acknowledgements = batch.getAcknowledgements();
                assertEquals(1, acknowledgements.size());
                assertEquals(AcknowledgeType.RELEASE, acknowledgements.get(2L));

                // Record 3 is returned in the next batch, because record 2 has now been skipped
                batch = completedFetch.fetchRecords(deserializers, 10, false);
                assertNull(batch.getException());
                fetchedRecords = batch.getInFlightRecords();
                assertEquals(1, fetchedRecords.size());
                assertEquals(3L, fetchedRecords.get(0).offset());
                acknowledgements = batch.getAcknowledgements();
                assertEquals(0, acknowledgements.size());
            }
        }
    }