@Override
    public OffsetStorageInfo loadOffset(String group, String topic, int partitionId) {
        String zkNode = new StringBuilder(512).append(this.consumerZkDir).append("/")
                .append(group).append("/offsets/").append(topic).append("/")
                .append(brokerId).append(TokenConstants.HYPHEN)
                .append(partitionId).toString();
        String offsetZkInfo;
        try {
            offsetZkInfo = ZKUtil.readDataMaybeNull(this.zkw, zkNode);
        } catch (KeeperException e) {
            BrokerSrvStatsHolder.incZKExcCnt();
            logger.error("KeeperException during load offsets from ZooKeeper", e);
            return null;
        }
        if (offsetZkInfo == null) {
            return null;
        }
        String[] offsetInfoStrs =
                offsetZkInfo.split(TokenConstants.HYPHEN);
        return new OffsetStorageInfo(topic, brokerId, partitionId, 0,
                Long.parseLong(offsetInfoStrs[1]), Long.parseLong(offsetInfoStrs[0]),
                false, System.currentTimeMillis());

    }
Initial ZooKeeper offset storage object

@param zkConfig   the ZooKeeper configure
@param isBroker   whether used in broker node
@param brokerId   the broker id
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/offsetstorage/ZkOffsetStorage.java