CompletableFuture<Map<TopicIdPartition, ShareFetchResponseData.PartitionData>> processFetchResponse(
        ShareFetchPartitionData shareFetchPartitionData,
        List<Tuple2<TopicIdPartition, FetchPartitionData>> responseData
    ) {
        Map<TopicIdPartition, CompletableFuture<PartitionData>> futures = new HashMap<>();
        responseData.forEach(data -> {
            TopicIdPartition topicIdPartition = data._1;
            FetchPartitionData fetchPartitionData = data._2;

            SharePartition sharePartition = partitionCacheMap.get(sharePartitionKey(shareFetchPartitionData.groupId, topicIdPartition));
            futures.put(topicIdPartition, sharePartition.acquire(shareFetchPartitionData.memberId, fetchPartitionData)
                .handle((acquiredRecords, throwable) -> {
                    log.trace("Acquired records for topicIdPartition: {} with share fetch data: {}, records: {}",
                        topicIdPartition, shareFetchPartitionData, acquiredRecords);
                    ShareFetchResponseData.PartitionData partitionData = new ShareFetchResponseData.PartitionData()
                        .setPartitionIndex(topicIdPartition.partition());

                    if (throwable != null) {
                        partitionData.setErrorCode(Errors.forException(throwable).code());
                        return partitionData;
                    }

                    if (fetchPartitionData.error.code() == Errors.OFFSET_OUT_OF_RANGE.code()) {
                        // In case we get OFFSET_OUT_OF_RANGE error, that's because the LSO is later than the fetch offset.
                        // So, we would update the start and end offset of the share partition and still return an empty
                        // response and let the client retry the fetch. This way we do not lose out on the data that
                        // would be returned for other share partitions in the fetch request.
                        sharePartition.updateCacheAndOffsets(offsetForEarliestTimestamp(topicIdPartition));
                        partitionData
                            .setPartitionIndex(topicIdPartition.partition())
                            .setRecords(null)
                            .setErrorCode(Errors.NONE.code())
                            .setAcquiredRecords(Collections.emptyList())
                            .setAcknowledgeErrorCode(Errors.NONE.code());
                        return partitionData;
                    }

                    // Maybe, in the future, check if no records are acquired, and we want to retry
                    // replica manager fetch. Depends on the share partition manager implementation,
                    // if we want parallel requests for the same share partition or not.
                    partitionData
                        .setPartitionIndex(topicIdPartition.partition())
                        .setRecords(fetchPartitionData.records)
                        .setErrorCode(fetchPartitionData.error.code())
                        .setAcquiredRecords(acquiredRecords)
                        .setAcknowledgeErrorCode(Errors.NONE.code());
                    return partitionData;
                }));
        });

        return CompletableFuture.allOf(futures.values().toArray(new CompletableFuture[0])).thenApply(v -> {
            Map<TopicIdPartition, ShareFetchResponseData.PartitionData> processedResult = new HashMap<>();
            futures.forEach((topicIdPartition, future) -> processedResult.put(topicIdPartition, future.join()));
            return processedResult;
        });
    }
Recursive function to process all the fetch requests present inside the fetch queue
core/src/main/java/kafka/server/share/SharePartitionManager.java