@Override
  public V chooseVolume(List<V> volumes, long replicaSize, String storageId)
      throws IOException {
    if (volumes.size() < 1) {
      throw new DiskOutOfSpaceException("No more available volumes");
    }
    // As all the items in volumes are with the same storage type,
    // so only need to get the storage type index of the first item in volumes
    StorageType storageType = volumes.get(0).getStorageType();
    int index = storageType != null ?
            storageType.ordinal() : StorageType.DEFAULT.ordinal();

    synchronized (syncLocks[index]) {
      return doChooseVolume(volumes, replicaSize, storageId);
    }
  }
@Test(timeout=60000)
  public void testFourUnbalancedVolumes() throws Exception {
    @SuppressWarnings("unchecked")
    final AvailableSpaceVolumeChoosingPolicy<FsVolumeSpi> policy = 
        ReflectionUtils.newInstance(AvailableSpaceVolumeChoosingPolicy.class, null);
    
    List<FsVolumeSpi> volumes = new ArrayList<FsVolumeSpi>();
    
    // First volume with 1MB free space
    volumes.add(Mockito.mock(FsVolumeSpi.class));
    Mockito.when(volumes.get(0).getAvailable()).thenReturn(1024L * 1024L);
    
    // Second volume with 1MB + 1 byte free space
    volumes.add(Mockito.mock(FsVolumeSpi.class));
    Mockito.when(volumes.get(1).getAvailable()).thenReturn(1024L * 1024L + 1);
    
    // Third volume with 3MB free space, which is a difference of 2MB, more
    // than the threshold of 1MB.
    volumes.add(Mockito.mock(FsVolumeSpi.class));
    Mockito.when(volumes.get(2).getAvailable()).thenReturn(1024L * 1024L * 3);
    
    // Fourth volume, again with 3MB free space.
    volumes.add(Mockito.mock(FsVolumeSpi.class));
    Mockito.when(volumes.get(3).getAvailable()).thenReturn(1024L * 1024L * 3);

    // We should alternate assigning between the two volumes with a lot of free
    // space.
    initPolicy(policy, BALANCED_SPACE_THRESHOLD, 1.0f);
    Assert.assertEquals(volumes.get(2), policy.chooseVolume(volumes, 100,
        null));
    Assert.assertEquals(volumes.get(3), policy.chooseVolume(volumes, 100,
        null));
    Assert.assertEquals(volumes.get(2), policy.chooseVolume(volumes, 100,
        null));
    Assert.assertEquals(volumes.get(3), policy.chooseVolume(volumes, 100,
        null));

    // We should alternate assigning between the two volumes with less free
    // space.
    initPolicy(policy, BALANCED_SPACE_THRESHOLD, 0.0f);
    Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100,
        null));
    Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100,
        null));
    Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100,
         null));
    Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100,
        null));
  }