private Optional<Throwable> releaseAcquiredRecordsForCompleteBatch(String memberId,
                                                                       InFlightBatch inFlightBatch,
                                                                       RecordState recordState,
                                                                       List<InFlightState> updatedStates,
                                                                       List<PersisterStateBatch> stateBatches) {

        // Check if member id is the owner of the batch.
        if (!inFlightBatch.batchMemberId().equals(memberId) && !inFlightBatch.batchMemberId().equals(EMPTY_MEMBER_ID)) {
            log.debug("Member {} is not the owner of batch record {} for share partition: {}-{}. Skipping batch.",
                    memberId, inFlightBatch, groupId, topicIdPartition);
            return Optional.empty();
        }

        // Change the state of complete batch since the same state exists for the entire inFlight batch.
        log.trace("Releasing acquired records for complete batch {} for the share partition: {}-{}",
                inFlightBatch, groupId, topicIdPartition);

        if (inFlightBatch.batchState() == RecordState.ACQUIRED) {
            InFlightState updateResult = inFlightBatch.startBatchStateTransition(
                    inFlightBatch.lastOffset() < startOffset ? RecordState.ARCHIVED : recordState,
                    false,
                    this.maxDeliveryCount,
                    EMPTY_MEMBER_ID
            );
            if (updateResult == null) {
                log.debug("Unable to release records from acquired state for the batch: {}"
                        + " for the share partition: {}-{}", inFlightBatch, groupId, topicIdPartition);
                return Optional.of(new InvalidRecordStateException("Unable to release acquired 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 next fetch offset because the record is not available for acquisition
            if (updateResult.state != RecordState.ARCHIVED) {
                findNextFetchOffset.set(true);
            }
        }
        return Optional.empty();
    }
Release the acquired records for the share partition. The next fetch offset is updated to the next offset
that should be fetched from the leader.

@param memberId The member id of the client whose records shall be released.

@return A future which is completed when the records are released.
core/src/main/java/kafka/server/share/SharePartition.java