private void initialize() {
        log.debug("Initializing share partition: {}-{}", groupId, topicIdPartition);
        // Initialize the share partition by reading the state from the persister.
        ReadShareGroupStateResult response;
        try {
            response = persister.readState(new ReadShareGroupStateParameters.Builder()
                .setGroupTopicPartitionData(new GroupTopicPartitionData.Builder<PartitionIdLeaderEpochData>()
                    .setGroupId(this.groupId)
                    .setTopicsData(Collections.singletonList(new TopicData<>(topicIdPartition.topicId(),
                        Collections.singletonList(PartitionFactory.newPartitionIdLeaderEpochData(topicIdPartition.partition(), 0)))))
                    .build())
                .build()
            ).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("Failed to initialize the share partition: {}-{}", groupId, topicIdPartition, e);
            throw new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition), e);
        }

        if (response == null || response.topicsData() == null || response.topicsData().size() != 1) {
            log.error("Failed to initialize the share partition: {}-{}. Invalid state found: {}.",
                groupId, topicIdPartition, response);
            throw new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
        }

        TopicData<PartitionAllData> state = response.topicsData().get(0);
        if (state.topicId() != topicIdPartition.topicId() || state.partitions().size() != 1
            || state.partitions().get(0).partition() != topicIdPartition.partition()) {
            log.error("Failed to initialize the share partition: {}-{}. Invalid topic partition response: {}.",
                groupId, topicIdPartition, response);
            throw new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
        }

        PartitionAllData partitionData = state.partitions().get(0);
        // Set the state epoch and end offset from the persisted state.
        startOffset = partitionData.startOffset() != -1 ? partitionData.startOffset() : 0;
        stateEpoch = partitionData.stateEpoch();

        List<PersisterStateBatch> stateBatches = partitionData.stateBatches();
        for (PersisterStateBatch stateBatch : stateBatches) {
            if (stateBatch.firstOffset() < startOffset) {
                log.error("Invalid state batch found for the share partition: {}-{}. The base offset: {}"
                        + " is less than the start offset: {}.", groupId, topicIdPartition,
                    stateBatch.firstOffset(), startOffset);
                throw new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
            }
            InFlightBatch inFlightBatch = new InFlightBatch(EMPTY_MEMBER_ID, stateBatch.firstOffset(),
                stateBatch.lastOffset(), RecordState.forId(stateBatch.deliveryState()), stateBatch.deliveryCount(), null);
            cachedState.put(stateBatch.firstOffset(), inFlightBatch);
        }
        // Update the endOffset of the partition.
        if (!cachedState.isEmpty()) {
            // If the cachedState is not empty, findNextFetchOffset flag is set to true so that any AVAILABLE records
            // in the cached state are not missed
            findNextFetchOffset.set(true);
            endOffset = cachedState.lastEntry().getValue().lastOffset();
            // In case the persister read state RPC result contains no AVAILABLE records, we can update cached state
            // and start/end offsets.
            maybeUpdateCachedStateAndOffsets();
        } else {
            endOffset = partitionData.startOffset();
        }
    }
@Test
    public void testInitialize() {
        Persister persister = Mockito.mock(Persister.class);
        ReadShareGroupStateResult readShareGroupStateResult = Mockito.mock(ReadShareGroupStateResult.class);
        Mockito.when(readShareGroupStateResult.topicsData()).thenReturn(Collections.singletonList(
            new TopicData<>(TOPIC_ID_PARTITION.topicId(), Collections.singletonList(
                PartitionFactory.newPartitionAllData(0, 3, 5L, Errors.NONE.code(), Errors.NONE.message(),
                    Arrays.asList(
                        new PersisterStateBatch(5L, 10L, RecordState.AVAILABLE.id, (short) 2),
                        new PersisterStateBatch(11L, 15L, RecordState.ARCHIVED.id, (short) 3)))))));
        Mockito.when(persister.readState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(readShareGroupStateResult));
        SharePartition sharePartition = SharePartitionBuilder.builder().withPersister(persister).build();

        assertFalse(sharePartition.cachedState().isEmpty());
        assertEquals(5, sharePartition.startOffset());
        assertEquals(15, sharePartition.endOffset());
        assertEquals(3, sharePartition.stateEpoch());
        assertEquals(5, sharePartition.nextFetchOffset());

        assertEquals(2, sharePartition.cachedState().size());
        assertNotNull(sharePartition.cachedState().get(5L));
        assertNotNull(sharePartition.cachedState().get(11L));

        assertEquals(10, sharePartition.cachedState().get(5L).lastOffset());
        assertEquals(RecordState.AVAILABLE, sharePartition.cachedState().get(5L).batchState());
        assertEquals(2, sharePartition.cachedState().get(5L).batchDeliveryCount());
        assertNull(sharePartition.cachedState().get(5L).offsetState());

        assertEquals(15, sharePartition.cachedState().get(11L).lastOffset());
        assertEquals(RecordState.ARCHIVED, sharePartition.cachedState().get(11L).batchState());
        assertEquals(3, sharePartition.cachedState().get(11L).batchDeliveryCount());
        assertNull(sharePartition.cachedState().get(11L).offsetState());
    }