ProvidedStorageMap(RwLock lock, BlockManager bm, Configuration conf) {

    storageId = conf.get(DFSConfigKeys.DFS_PROVIDER_STORAGEUUID,
        DFSConfigKeys.DFS_PROVIDER_STORAGEUUID_DEFAULT);

    providedEnabled = conf.getBoolean(
        DFSConfigKeys.DFS_NAMENODE_PROVIDED_ENABLED,
        DFSConfigKeys.DFS_NAMENODE_PROVIDED_ENABLED_DEFAULT);

    if (!providedEnabled) {
      // disable mapping
      aliasMap = null;
      providedDescriptor = null;
      providedStorageInfo = null;
      return;
    }

    DatanodeStorage ds = new DatanodeStorage(
        storageId, State.NORMAL, StorageType.PROVIDED);
    providedDescriptor = new ProvidedDescriptor();
    providedStorageInfo = providedDescriptor.createProvidedStorage(ds);
    this.defaultReplication = conf.getInt(DFSConfigKeys.DFS_REPLICATION_KEY,
        DFSConfigKeys.DFS_REPLICATION_DEFAULT);

    this.bm = bm;
    this.lock = lock;

    // load block reader into storage
    Class<? extends BlockAliasMap> aliasMapClass = conf.getClass(
            DFSConfigKeys.DFS_PROVIDED_ALIASMAP_CLASS,
            TextFileRegionAliasMap.class, BlockAliasMap.class);
    aliasMap = ReflectionUtils.newInstance(aliasMapClass, conf);

    LOG.info("Loaded alias map class: " +
        aliasMap.getClass() + " storage: " + providedStorageInfo);
  }
@Test
  public void testProvidedStorageMap() throws IOException {
    ProvidedStorageMap providedMap = new ProvidedStorageMap(
        nameSystemLock, bm, conf);
    DatanodeStorageInfo providedMapStorage =
        providedMap.getProvidedStorageInfo();
    // the provided storage cannot be null
    assertNotNull(providedMapStorage);

    // create a datanode
    DatanodeDescriptor dn1 = createDatanodeDescriptor(5000);

    // associate two storages to the datanode
    DatanodeStorage dn1ProvidedStorage = new DatanodeStorage(
        providedStorageID,
        DatanodeStorage.State.NORMAL,
        StorageType.PROVIDED);
    DatanodeStorage dn1DiskStorage = new DatanodeStorage(
        "sid-1", DatanodeStorage.State.NORMAL, StorageType.DISK);

    when(nameSystemLock.hasWriteLock()).thenReturn(true);
    DatanodeStorageInfo dns1Provided =
        providedMap.getStorage(dn1, dn1ProvidedStorage);
    DatanodeStorageInfo dns1Disk = providedMap.getStorage(dn1, dn1DiskStorage);

    assertTrue("The provided storages should be equal",
        dns1Provided == providedMapStorage);
    assertTrue("Disk storage has not yet been registered with block manager",
        dns1Disk == null);
    // add the disk storage to the datanode.
    DatanodeStorageInfo dnsDisk = new DatanodeStorageInfo(dn1, dn1DiskStorage);
    dn1.injectStorage(dnsDisk);
    assertTrue("Disk storage must match the injected storage info",
        dnsDisk == providedMap.getStorage(dn1, dn1DiskStorage));

    // create a 2nd datanode
    DatanodeDescriptor dn2 = createDatanodeDescriptor(5010);
    // associate a provided storage with the datanode
    DatanodeStorage dn2ProvidedStorage = new DatanodeStorage(
        providedStorageID,
        DatanodeStorage.State.NORMAL,
        StorageType.PROVIDED);

    DatanodeStorageInfo dns2Provided = providedMap.getStorage(
        dn2, dn2ProvidedStorage);
    assertTrue("The provided storages should be equal",
        dns2Provided == providedMapStorage);
    assertTrue("The DatanodeDescriptor should contain the provided storage",
        dn2.getStorageInfo(providedStorageID) == providedMapStorage);
  }