void rollbackOrProcessStateUpdates(
        Throwable throwable,
        List<InFlightState> updatedStates,
        List<PersisterStateBatch> stateBatches
    ) {
        if (stateBatches.isEmpty() && updatedStates.isEmpty())
            return;

        lock.writeLock().lock();
        try {
            if (throwable != null || !isWriteShareGroupStateSuccessful(stateBatches)) {
                // Log in DEBUG to avoid flooding of logs for a faulty client.
                log.debug("Request failed for updating state, rollback any changed state"
                    + " for the share partition: {}-{}", groupId, topicIdPartition);
                updatedStates.forEach(state -> state.completeStateTransition(false));
            } else {
                log.trace("State change request successful for share partition: {}-{}",
                    groupId, topicIdPartition);
                updatedStates.forEach(state -> {
                    state.completeStateTransition(true);
                    // Cancel the acquisition lock timeout task for the state since it is acknowledged/released successfully.
                    state.cancelAndClearAcquisitionLockTimeoutTask();
                });
                // Update the cached state and start and end offsets after acknowledging/releasing the acquired records.
                maybeUpdateCachedStateAndOffsets();
            }
        } 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