private NavigableMap<Long, InFlightBatch> fetchSubMapForAcknowledgementBatch(
        ShareAcknowledgementBatch batch,
        boolean isLastBatch
    ) {
        lock.writeLock().lock();
        try {
            // Find the floor batch record for the request batch. The request batch could be
            // for a subset of the batch i.e. cached batch of offset 10-14 and request batch
            // of 12-13. Hence, floor entry is fetched to find the sub-map.
            Map.Entry<Long, InFlightBatch> floorOffset = cachedState.floorEntry(batch.firstOffset());
            if (floorOffset == null) {
                boolean hasStartOffsetMoved = checkForStartOffsetWithinBatch(batch.firstOffset(), batch.lastOffset());
                if (hasStartOffsetMoved) {
                    // If the start offset has been moved and within the request batch then fetch
                    // the floor entry from start offset and acknowledge cached offsets. Consider
                    // the case where the start offset has moved from 0 to 10, with the cached batch
                    // of 0 - 5, 5 - 10, 10 - 12, 12 - 15. The request batch for acknowledgement is 5 - 15,
                    // then post acquisition lock timeout the cache will have data from only from 10 to 15.
                    // Hence, we need to fetch the floor entry from start offset.
                    floorOffset = cachedState.floorEntry(startOffset);
                } else {
                    log.debug("Batch record {} not found for share partition: {}-{}", batch, groupId,
                        topicIdPartition);
                    throw new InvalidRecordStateException(
                        "Batch record not found. The request batch offsets are not found in the cache.");
                }
            }

            NavigableMap<Long, InFlightBatch> subMap = cachedState.subMap(floorOffset.getKey(), true, batch.lastOffset(), true);
            // Validate if the request batch has the first offset less than the last offset of the last
            // fetched cached batch, then there will be offsets in the request that are already acquired.
            if (subMap.lastEntry().getValue().lastOffset < batch.firstOffset()) {
                log.debug("Request batch: {} has offsets which are not found for share partition: {}-{}", batch, groupId, topicIdPartition);
                throw new InvalidRequestException("Batch record not found. The first offset in request is past acquired records.");
            }

            // Validate if the request batch has the last offset greater than the last offset of
            // the last fetched cached batch, then there will be offsets in the request than cannot
            // be found in the fetched batches.
            if (isLastBatch && batch.lastOffset() > subMap.lastEntry().getValue().lastOffset) {
                log.debug("Request batch: {} has offsets which are not found for share partition: {}-{}", batch, groupId, topicIdPartition);
                throw new InvalidRequestException("Batch record not found. The last offset in request is past acquired records.");
            }

            return subMap;
        } finally {
            lock.writeLock().unlock();
        }
    }
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