private <K, V> void rejectRecordBatch(final ShareInFlightBatch<K, V> inFlightBatch,
                                          final RecordBatch currentBatch) {
        // Rewind the acquiredRecordIterator to the start, so we are in a known state
        acquiredRecordIterator = acquiredRecordList.listIterator();

        OffsetAndDeliveryCount nextAcquired = nextAcquiredRecord();
        for (long offset = currentBatch.baseOffset(); offset <= currentBatch.lastOffset(); offset++) {
            if (nextAcquired == null) {
                // No more acquired records, so we are done
                break;
            } else if (offset == nextAcquired.offset) {
                // It's acquired, so we reject it
                inFlightBatch.addAcknowledgement(offset, AcknowledgeType.REJECT);
            } else if (offset < nextAcquired.offset) {
                // It's not acquired, so we skip it
                continue;
            }

            nextAcquired = nextAcquiredRecord();
        }
    }
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