private Optional<Throwable> acknowledgeCompleteBatch(
        ShareAcknowledgementBatch batch,
        InFlightBatch inFlightBatch,
        RecordState recordState,
        List<InFlightState> updatedStates,
        List<PersisterStateBatch> stateBatches
    ) {
        lock.writeLock().lock();
        try {
            // The in-flight batch is a full match hence change the state of the complete.
            log.trace("Acknowledging complete batch record {} for the share partition: {}-{}",
                batch, groupId, topicIdPartition);
            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 batch is not in the acquired state."));
            }

            // Change the state of complete batch since the same state exists for the entire inFlight batch.
            // The member id is reset to EMPTY_MEMBER_ID irrespective of the acknowledge type as the batch is
            // either released or moved to a state where member id existence is not important. The member id
            // is only important when the batch is acquired.
            InFlightState updateResult = inFlightBatch.startBatchStateTransition(
                recordState,
                false,
                this.maxDeliveryCount,
                EMPTY_MEMBER_ID
            );
            if (updateResult == null) {
                log.debug("Unable to acknowledge records for the batch: {} with state: {}"
                        + " for the share partition: {}-{}", inFlightBatch, recordState, groupId,
                    topicIdPartition);
                return Optional.of(
                    new InvalidRecordStateException("Unable to acknowledge records for the batch"));
            }

            // Successfully updated the state of the batch.
            updatedStates.add(updateResult);
            stateBatches.add(
                new PersisterStateBatch(inFlightBatch.firstOffset, inFlightBatch.lastOffset,
                    updateResult.state.id, (short) updateResult.deliveryCount));

            // If the maxDeliveryCount limit has been exceeded, the record will be transitioned to ARCHIVED state.
            // This should not change the nextFetchOffset because the record is not available for acquisition
            if (recordState == RecordState.AVAILABLE
                && updateResult.state != RecordState.ARCHIVED) {
                findNextFetchOffset.set(true);
            }
        } 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