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

        log.trace("Offset tracked batch record found, batch: {} for the share partition: {}-{}", inFlightBatch,
                groupId, topicIdPartition);
        for (Map.Entry<Long, InFlightState> offsetState : inFlightBatch.offsetState.entrySet()) {

            // Check if member id is the owner of the offset.
            if (!offsetState.getValue().memberId().equals(memberId) && !offsetState.getValue().memberId().equals(EMPTY_MEMBER_ID)) {
                log.debug("Member {} is not the owner of offset: {} in batch: {} for the share"
                        + " partition: {}-{}. Skipping offset.", memberId, offsetState.getKey(), inFlightBatch, groupId, topicIdPartition);
                return Optional.empty();
            }
            if (offsetState.getValue().state == RecordState.ACQUIRED) {
                InFlightState updateResult = offsetState.getValue().startStateTransition(
                        offsetState.getKey() < startOffset ? RecordState.ARCHIVED : recordState,
                        false,
                        this.maxDeliveryCount,
                        EMPTY_MEMBER_ID
                );
                if (updateResult == null) {
                    log.debug("Unable to release records from acquired state for the offset: {} in batch: {}"
                                    + " for the share partition: {}-{}", offsetState.getKey(),
                            inFlightBatch, groupId, topicIdPartition);
                    return Optional.of(new InvalidRecordStateException("Unable to release acquired records for the offset"));
                }

                // Successfully updated the state of the offset.
                updatedStates.add(updateResult);
                stateBatches.add(new PersisterStateBatch(offsetState.getKey(), offsetState.getKey(),
                        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