public CompletableFuture<Optional<Throwable>> acknowledge(
        String memberId,
        List<ShareAcknowledgementBatch> acknowledgementBatch
    ) {
        log.trace("Acknowledgement batch request for share partition: {}-{}", groupId, topicIdPartition);

        Throwable throwable = null;
        lock.writeLock().lock();
        List<InFlightState> updatedStates = new ArrayList<>();
        List<PersisterStateBatch> stateBatches = new ArrayList<>();
        try {
            // Avoided using enhanced for loop as need to check if the last batch have offsets
            // in the range.
            for (int i = 0; i < acknowledgementBatch.size(); i++) {
                ShareAcknowledgementBatch batch = acknowledgementBatch.get(i);

                // Client can either send a single entry in acknowledgeTypes which represents the state
                // of the complete batch or can send individual offsets state.
                Map<Long, RecordState> recordStateMap;
                try {
                    recordStateMap = fetchRecordStateMapForAcknowledgementBatch(batch);
                } catch (IllegalArgumentException e) {
                    log.debug("Invalid acknowledge type: {} for share partition: {}-{}",
                        batch.acknowledgeTypes(), groupId, topicIdPartition);
                    throwable = new InvalidRequestException("Invalid acknowledge type: " + batch.acknowledgeTypes());
                    break;
                }

                if (batch.lastOffset() < startOffset) {
                    log.trace("All offsets in the acknowledgement batch {} are already archived: {}-{}",
                        batch, groupId, topicIdPartition);
                    continue;
                }

                // Fetch the sub-map from the cached map for the batch to acknowledge. The sub-map can
                // be a full match, subset or spans over multiple fetched batches.
                NavigableMap<Long, InFlightBatch> subMap;
                try {
                    subMap = fetchSubMapForAcknowledgementBatch(batch, i == acknowledgementBatch.size() - 1);
                } catch (InvalidRecordStateException | InvalidRequestException e) {
                    throwable = e;
                    break;
                }

                // Acknowledge the records for the batch.
                Optional<Throwable> ackThrowable = acknowledgementBatchRecords(
                    memberId,
                    batch,
                    recordStateMap,
                    subMap,
                    updatedStates,
                    stateBatches
                );

                if (ackThrowable.isPresent()) {
                    throwable = ackThrowable.get();
                    break;
                }
            }

            // If the acknowledgement is successful then persist state, complete the state transition
            // and update the cached state for start offset. Else rollback the state transition.
            rollbackOrProcessStateUpdates(throwable, updatedStates, stateBatches);
        } finally {
            lock.writeLock().unlock();
        }

        return CompletableFuture.completedFuture(Optional.ofNullable(throwable));
    }
Acknowledge the fetched records for the share partition. The accepted batches are removed
from the in-flight records once persisted. The next fetch offset is updated to the next offset
that should be fetched from the leader, if required.

@param memberId The member id of the client that is fetching the record.
@param acknowledgementBatch The acknowledgement batch list for the share partition.

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