code: private UnderFileSystem getOrAdd(AlluxioURI ufsUri, UnderFileSystemConfiguration ufsConf) {
      Key key = new Key(ufsUri, ufsConf.getMountSpecificConf());
      UnderFileSystem cachedFs = mUnderFileSystemMap.get(key);
      if (cachedFs != null) {
        return cachedFs;
      }
      // On cache miss, synchronize the creation to ensure ufs is only created once
      synchronized (mLock) {
        cachedFs = mUnderFileSystemMap.get(key);
        if (cachedFs != null) {
          return cachedFs;
        }
        UnderFileSystem fs = UnderFileSystem.Factory.create(ufsUri.toString(), ufsConf);
  
        // Detect whether to use managed blocking on UFS operations.
        boolean useManagedBlocking = fs.isObjectStorage();
        if (ufsConf.isSet(PropertyKey.UNDERFS_RUN_WITH_MANAGEDBLOCKING)) {
          useManagedBlocking = ufsConf.getBoolean(PropertyKey.UNDERFS_RUN_WITH_MANAGEDBLOCKING);
        }
        // Wrap UFS under managed blocking forwarder if required.
        if (useManagedBlocking) {
          fs = new ManagedBlockingUfsForwarder(fs);
        }
  
        mUnderFileSystemMap.putIfAbsent(key, fs);
        mCloser.register(fs);
        try {
          connectUfs(fs);
        } catch (IOException e) {
          LOG.warn("Failed to perform initial connect to UFS {}: {}", ufsUri, e.getMessage());
        }
        return fs;
      }
    }
docs: Return a UFS instance if it already exists in the cache, otherwise, creates a new instance and
  return this.
  
  @param ufsUri the UFS path
  @param ufsConf the UFS configuration
  @return the UFS instance