CompletableFuture<Map<TopicIdPartition, 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);
                    PartitionData partitionData = new 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, PartitionData> processedResult = new HashMap<>();
            futures.forEach((topicIdPartition, future) -> processedResult.put(topicIdPartition, future.join()));
            return processedResult;
        });
    }
@Test
    public void testProcessFetchResponse() {
        String groupId = "grp";
        String memberId = Uuid.randomUuid().toString();
        Uuid topicId = Uuid.randomUuid();
        TopicIdPartition tp0 = new TopicIdPartition(topicId, new TopicPartition("foo", 0));
        TopicIdPartition tp1 = new TopicIdPartition(topicId, new TopicPartition("foo", 1));
        Map<TopicIdPartition, Integer> partitionMaxBytes = new HashMap<>();
        partitionMaxBytes.put(tp0, PARTITION_MAX_BYTES);
        partitionMaxBytes.put(tp1, PARTITION_MAX_BYTES);

        Map<SharePartitionManager.SharePartitionKey, SharePartition> partitionCacheMap = new ConcurrentHashMap<>();
        partitionCacheMap.computeIfAbsent(new SharePartitionManager.SharePartitionKey(groupId, tp0),
            k -> new SharePartition(groupId, tp0, MAX_IN_FLIGHT_MESSAGES, MAX_DELIVERY_COUNT,
                RECORD_LOCK_DURATION_MS, mockTimer, new MockTime(), NoOpShareStatePersister.getInstance()));
        partitionCacheMap.computeIfAbsent(new SharePartitionManager.SharePartitionKey(groupId, tp1),
            k -> new SharePartition(groupId, tp1, MAX_IN_FLIGHT_MESSAGES, MAX_DELIVERY_COUNT,
                RECORD_LOCK_DURATION_MS, mockTimer, new MockTime(), NoOpShareStatePersister.getInstance()));

        SharePartitionManager sharePartitionManager = SharePartitionManagerBuilder.builder()
            .withPartitionCacheMap(partitionCacheMap).build();

        CompletableFuture<Map<TopicIdPartition, PartitionData>> future = new CompletableFuture<>();
        SharePartitionManager.ShareFetchPartitionData shareFetchPartitionData = new SharePartitionManager.ShareFetchPartitionData(
            new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, 0,
                1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, memberId,
                future, partitionMaxBytes);

        MemoryRecords records = MemoryRecords.withRecords(Compression.NONE,
            new SimpleRecord("0".getBytes(), "v".getBytes()),
            new SimpleRecord("1".getBytes(), "v".getBytes()),
            new SimpleRecord("2".getBytes(), "v".getBytes()),
            new SimpleRecord(null, "value".getBytes()));

        MemoryRecords records1 = MemoryRecords.withRecords(100L, Compression.NONE,
            new SimpleRecord("0".getBytes(), "v".getBytes()),
            new SimpleRecord("1".getBytes(), "v".getBytes()),
            new SimpleRecord("2".getBytes(), "v".getBytes()),
            new SimpleRecord(null, "value".getBytes()));

        List<Tuple2<TopicIdPartition, FetchPartitionData>> responseData = new ArrayList<>();
        responseData.add(new Tuple2<>(tp0, new FetchPartitionData(Errors.NONE, 0L, 0L,
            records, Optional.empty(), OptionalLong.empty(), Optional.empty(),
            OptionalInt.empty(), false)));
        responseData.add(new Tuple2<>(tp1, new FetchPartitionData(Errors.NONE, 0L, 100L,
            records1, Optional.empty(), OptionalLong.empty(), Optional.empty(),
            OptionalInt.empty(), false)));
        CompletableFuture<Map<TopicIdPartition, ShareFetchResponseData.PartitionData>> result =
            sharePartitionManager.processFetchResponse(shareFetchPartitionData, responseData);

        assertTrue(result.isDone());
        Map<TopicIdPartition, ShareFetchResponseData.PartitionData> resultData = result.join();
        assertEquals(2, resultData.size());
        assertTrue(resultData.containsKey(tp0));
        assertTrue(resultData.containsKey(tp1));
        assertEquals(0, resultData.get(tp0).partitionIndex());
        assertEquals(1, resultData.get(tp1).partitionIndex());
        assertEquals(Errors.NONE.code(), resultData.get(tp0).errorCode());
        assertEquals(Errors.NONE.code(), resultData.get(tp1).errorCode());
        assertEquals(Collections.singletonList(new ShareFetchResponseData.AcquiredRecords()
                .setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)),
            resultData.get(tp0).acquiredRecords());
        assertEquals(Collections.singletonList(new ShareFetchResponseData.AcquiredRecords()
                .setFirstOffset(100).setLastOffset(103).setDeliveryCount((short) 1)),
            resultData.get(tp1).acquiredRecords());
    }