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();
        }
    }
Release the fetch lock once the records are fetched from the leader.
core/src/main/java/kafka/server/share/SharePartition.java