private void maybeUpdateCachedStateAndOffsets() {
        lock.writeLock().lock();
        try {
            if (!canMoveStartOffset()) {
                return;
            }

            // This will help to find the next position for the startOffset.
            // The new position of startOffset will be lastOffsetAcknowledged + 1
            long lastOffsetAcknowledged = findLastOffsetAcknowledged();
            // If lastOffsetAcknowledged is -1, this means we cannot move out startOffset ahead
            if (lastOffsetAcknowledged == -1) {
                return;
            }

            // This is true if all records in the cachedState have been acknowledged (either Accept or Reject).
            // The resulting action should be to empty the cachedState altogether
            long lastCachedOffset = cachedState.lastEntry().getValue().lastOffset();
            if (lastOffsetAcknowledged == lastCachedOffset) {
                startOffset = lastCachedOffset + 1; // The next offset that will be fetched and acquired in the share partition
                endOffset = lastCachedOffset + 1;
                cachedState.clear();
                // Nothing further to do.
                return;
            }

            /*
             The cachedState contains some records that are yet to be acknowledged, and thus should
             not be removed. Only a subMap will be removed from the cachedState. The logic to remove
             batches from cachedState is as follows:
             a) Only full batches can be removed from the cachedState, For example if there is batch (0-99)
             and 0-49 records are acknowledged (ACCEPT or REJECT), the first 50 records will not be removed
             from the cachedState. Instead, the startOffset will be moved to 50, but the batch will only
             be removed once all the messages (0-99) are acknowledged (ACCEPT or REJECT).
            */

            // Since only a subMap will be removed, we need to find the first and last keys of that subMap
            long firstKeyToRemove = cachedState.firstKey();
            long lastKeyToRemove;
            NavigableMap.Entry<Long, InFlightBatch> entry = cachedState.floorEntry(lastOffsetAcknowledged);
            if (lastOffsetAcknowledged == entry.getValue().lastOffset()) {
                startOffset = cachedState.higherKey(lastOffsetAcknowledged);
                lastKeyToRemove = entry.getKey();
            } else {
                startOffset = lastOffsetAcknowledged + 1;
                if (entry.getKey().equals(cachedState.firstKey())) {
                    // If the first batch in cachedState has some records yet to be acknowledged,
                    // then nothing should be removed from cachedState
                    lastKeyToRemove = -1;
                } else {
                    lastKeyToRemove = cachedState.lowerKey(entry.getKey());
                }
            }

            if (lastKeyToRemove != -1) {
                NavigableMap<Long, InFlightBatch> subMap = cachedState.subMap(firstKeyToRemove, true, lastKeyToRemove, true);
                for (Long key : subMap.keySet()) {
                    cachedState.remove(key);
                }
            }
        } 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