private boolean archivePerOffsetBatchRecords(InFlightBatch inFlightBatch,
                                                 long startOffsetToArchive,
                                                 long endOffsetToArchive) {
        lock.writeLock().lock();
        try {
            boolean isAnyOffsetArchived = false;
            log.trace("Archiving offset tracked batch: {} for the share partition: {}-{}", inFlightBatch, groupId, topicIdPartition);
            for (Map.Entry<Long, InFlightState> offsetState : inFlightBatch.offsetState().entrySet()) {
                if (offsetState.getKey() < startOffsetToArchive) {
                    continue;
                }
                if (offsetState.getKey() > endOffsetToArchive) {
                    // No further offsets to process.
                    break;
                }
                if (offsetState.getValue().state != RecordState.AVAILABLE) {
                    continue;
                }

                offsetState.getValue().archive(EMPTY_MEMBER_ID);
                isAnyOffsetArchived = true;
            }
            return isAnyOffsetArchived;
        } finally {
            lock.writeLock().unlock();
        }
    }
Updates the cached state, start and end offsets of the share partition as per the new log
start offset. The method is called when the log start offset is moved for the share partition.

@param logStartOffset The new log start offset.
core/src/main/java/kafka/server/share/SharePartition.java