private Optional<Throwable> acknowledgePerOffsetBatchRecords(
        String memberId,
        ShareAcknowledgementBatch batch,
        InFlightBatch inFlightBatch,
        Map<Long, RecordState> recordStateMap,
        List<InFlightState> updatedStates,
        List<PersisterStateBatch> stateBatches
    ) {
        lock.writeLock().lock();
        try {
            // Fetch the first record state from the map to be used as default record state in case the
            // offset record state is not provided by client.
            RecordState recordStateDefault = recordStateMap.get(batch.firstOffset());
            for (Map.Entry<Long, InFlightState> offsetState : inFlightBatch.offsetState.entrySet()) {

                // 1. For the first batch which might have offsets prior to the request base
                // offset i.e. cached batch of 10-14 offsets and request batch of 12-13.
                // 2. Skip the offsets which are below the start offset of the share partition
                if (offsetState.getKey() < batch.firstOffset() || offsetState.getKey() < startOffset) {
                    continue;
                }

                if (offsetState.getKey() > batch.lastOffset()) {
                    // No further offsets to process.
                    break;
                }

                if (offsetState.getValue().state != RecordState.ACQUIRED) {
                    log.debug("The offset is not acquired, offset: {} batch: {} for the share"
                            + " partition: {}-{}", offsetState.getKey(), inFlightBatch, groupId,
                        topicIdPartition);
                    return Optional.of(new InvalidRecordStateException(
                        "The batch cannot be acknowledged. The offset is not acquired."));
                }

                // Check if member id is the owner of the offset.
                if (!offsetState.getValue().memberId.equals(memberId)) {
                    log.debug("Member {} is not the owner of offset: {} in batch: {} for the share"
                            + " partition: {}-{}", memberId, offsetState.getKey(), inFlightBatch,
                        groupId, topicIdPartition);
                    return Optional.of(
                        new InvalidRecordStateException("Member is not the owner of offset"));
                }

                // Determine the record state for the offset. If the per offset record state is not provided
                // by the client, then use the batch record state.
                RecordState recordState =
                    recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
                        recordStateDefault;
                InFlightState updateResult = offsetState.getValue().startStateTransition(
                    recordState,
                    false,
                    this.maxDeliveryCount,
                    EMPTY_MEMBER_ID
                );
                if (updateResult == null) {
                    log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
                            + " for the share partition: {}-{}", offsetState.getKey(),
                        inFlightBatch, groupId, topicIdPartition);
                    return Optional.of(new InvalidRecordStateException(
                        "Unable to acknowledge records for the batch"));
                }
                // Successfully updated the state of the offset.
                updatedStates.add(updateResult);
                stateBatches.add(new PersisterStateBatch(offsetState.getKey(), offsetState.getKey(),
                    updateResult.state.id, (short) updateResult.deliveryCount));
                // If the maxDeliveryCount limit has been exceeded, the record will be transitioned to ARCHIVED state.
                // This should not change the next fetch offset because the record is not available for acquisition
                if (recordState == RecordState.AVAILABLE
                    && updateResult.state != RecordState.ARCHIVED) {
                    findNextFetchOffset.set(true);
                }
            }
        } finally {
            lock.writeLock().unlock();
        }
        return Optional.empty();
    }
Check if the start offset has moved and within the request first and last offset.

@param batchFirstOffset The first offset of the batch.
@param batchLastOffset The last offset of the batch.

@return True if the start offset has moved and within the request first and last offset, false otherwise.
core/src/main/java/kafka/server/share/SharePartition.java