private Optional<Throwable> acknowledgementBatchRecords(
        String memberId,
        ShareAcknowledgementBatch batch,
        Map<Long, RecordState> recordStateMap,
        NavigableMap<Long, InFlightBatch> subMap,
        final List<InFlightState> updatedStates,
        List<PersisterStateBatch> stateBatches
    ) {
        Optional<Throwable> throwable;
        lock.writeLock().lock();
        try {
            // The acknowledgement batch either is exact fetch equivalent batch (mostly), subset
            // or spans over multiple fetched batches. The state can vary per offset itself from
            // the fetched batch in case of subset or client sent individual offsets state.
            for (Map.Entry<Long, InFlightBatch> entry : subMap.entrySet()) {
                InFlightBatch inFlightBatch = entry.getValue();

                // If startOffset has moved ahead of the in-flight batch, skip the batch.
                if (inFlightBatch.lastOffset() < startOffset) {
                    log.trace("All offsets in the inflight batch {} are already archived: {}-{}",
                        inFlightBatch, groupId, topicIdPartition);
                    continue;
                }

                // Validate if the requested member id is the owner of the batch.
                if (inFlightBatch.offsetState() == null) {
                    throwable = validateAcknowledgementBatchMemberId(memberId, inFlightBatch);
                    if (throwable.isPresent()) {
                        return throwable;
                    }
                }

                // Determine if the in-flight batch is a full match from the request batch.
                boolean fullMatch = checkForFullMatch(inFlightBatch, batch.firstOffset(), batch.lastOffset());
                boolean isPerOffsetClientAck = batch.acknowledgeTypes().size() > 1;
                boolean hasStartOffsetMoved = checkForStartOffsetWithinBatch(inFlightBatch.firstOffset(), inFlightBatch.lastOffset());

                // Maintain state per offset if the inflight batch is not a full match or the
                // offset state is managed or client sent individual offsets state or
                // the start offset is within this in-flight batch.
                if (!fullMatch || inFlightBatch.offsetState() != null || isPerOffsetClientAck || hasStartOffsetMoved) {
                    log.debug("Subset or offset tracked batch record found for acknowledgement,"
                            + " batch: {}, request offsets - first: {}, last: {}, client per offset"
                            + "state {} for the share partition: {}-{}", inFlightBatch, batch.firstOffset(),
                        batch.lastOffset(), isPerOffsetClientAck, groupId, topicIdPartition);
                    if (inFlightBatch.offsetState() == null) {
                        // Though the request is a subset of in-flight batch but the offset
                        // tracking has not been initialized yet which means that we could only
                        // acknowledge subset of offsets from the in-flight batch but only if the
                        // complete batch is acquired yet. Hence, do a pre-check to avoid exploding
                        // the in-flight offset tracking unnecessarily.
                        if (inFlightBatch.batchState() != RecordState.ACQUIRED) {
                            log.debug("The batch is not in the acquired state: {} for share partition: {}-{}",
                                inFlightBatch, groupId, topicIdPartition);
                            return Optional.of(new InvalidRecordStateException("The batch cannot be acknowledged. The subset batch is not in the acquired state."));
                        }
                        // The request batch is a subset or per offset state is managed hence update
                        // the offsets state in the in-flight batch.
                        inFlightBatch.maybeInitializeOffsetStateUpdate();
                    }

                    throwable = acknowledgePerOffsetBatchRecords(memberId, batch, inFlightBatch,
                        recordStateMap, updatedStates, stateBatches);
                } else {
                    // The in-flight batch is a full match hence change the state of the complete batch.
                    throwable = acknowledgeCompleteBatch(batch, inFlightBatch,
                        recordStateMap.get(batch.firstOffset()), updatedStates, stateBatches);
                }

                if (throwable.isPresent()) {
                    return throwable;
                }
            }
        } 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