private GroupOffsetStgInfo loadZkOffsetStgInfo() {
        Set<String> tmpSet;
        Set<String> topicSet;
        TopicMetadata topicMeta;
        Map<String, Set<String>> zkAllGroups =
                zkOffsetStorage.queryGroupTopicInfo(null);
        // filter expired records
        Map<String, Set<String>> notFoundMap = new HashMap<>();
        for (Map.Entry<String, Set<String>> entry : zkAllGroups.entrySet()) {
            for (String topicName : entry.getValue()) {
                topicMeta = metadataManager.getTopicMetadata(topicName);
                if (topicMeta == null || !topicMeta.isValidTopic()) {
                    topicSet = notFoundMap.get(entry.getKey());
                    if (topicSet == null) {
                        tmpSet = new HashSet<>();
                        topicSet = notFoundMap.putIfAbsent(entry.getKey(), tmpSet);
                        if (topicSet == null) {
                            topicSet = tmpSet;
                        }
                    }
                    topicSet.add(topicName);
                }
            }
        }
        for (Map.Entry<String, Set<String>> entry : notFoundMap.entrySet()) {
            topicSet = zkAllGroups.get(entry.getKey());
            topicSet.removeAll(entry.getValue());
            if (topicSet.isEmpty()) {
                zkAllGroups.remove(entry.getKey());
            }
        }
        // get stored offset values
        Set<Integer> partIdSet;
        OffsetStorageInfo tmpRegInfo;
        GroupOffsetStgInfo tmpStgInfo = new GroupOffsetStgInfo(brokerConfig.getBrokerId());
        for (Map.Entry<String, Set<String>> entry : zkAllGroups.entrySet()) {
            if (this.isStopped()) {
                logger.info("[Offset Manager] found service stopped, exit 1!");
                return null;
            }
            topicSet = entry.getValue();
            for (String topicName : topicSet) {
                if (this.isStopped()) {
                    logger.info("[Offset Manager] found service stopped, exit 2!");
                    return null;
                }
                topicMeta = metadataManager.getTopicMetadata(topicName);
                if (topicMeta == null || !topicMeta.isValidTopic()) {
                    continue;
                }
                partIdSet = topicMeta.getAllPartitionIds();
                for (Integer partId : partIdSet) {
                    if (this.isStopped()) {
                        logger.info("[Offset Manager] found service stopped, exit 3!");
                        return null;
                    }
                    tmpRegInfo = zkOffsetStorage.loadOffset(entry.getKey(), topicName, partId);
                    if (tmpRegInfo == null) {
                        continue;
                    }
                    tmpStgInfo.addOffsetStgInfo(entry.getKey(), topicName, partId,
                            tmpRegInfo.getOffset(), tmpRegInfo.getMessageId());
                }
            }
        }
        return tmpStgInfo;
    }
Load or create offset.

@param group            the consume group name
@param topic            the consumed topic name
@param partitionId      the consumed partition id
@param offsetCacheKey   the offset store key
@param defOffset        the default offset if not found
@return                 the stored offset object
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/DefaultOffsetManager.java