@Test
  public void skipBadDiskTest() throws Exception {
    // clean up
    for (String mountPath : mountPathToReplicaTokenInfos.keySet()) {
      File[] files = (new File(mountPath)).listFiles();
      if (files != null) {
        for (File file : files) {
          Utils.deleteFileOrDirectory(file);
        }
      }
    }
    DiskTokenPersistor diskTokenPersistor = new DiskTokenPersistor(REPLICA_TOKEN_FILENAME, mountPathToPartitionInfoList,
        new ReplicationMetrics(new MetricRegistry(), Collections.emptyList()), clusterMap, findTokenHelper,
        mockStoreManager);
    // mock I/O exception for 1st mount path and all stores are down on that disk (Disk_Unavailable)
    Iterator<String> pathItor = mountPathToReplicaTokenInfos.keySet().iterator();
    String pathWithException1 = pathItor.next();
    File mountPathDir1 = new File(pathWithException1);
    assertTrue("Can't make dir unwritable", mountPathDir1.setWritable(false));
    Mockito.when(mockStoreManager.checkLocalPartitionStatus(any(), any())).thenReturn(ServerErrorCode.DiskUnavailable);
    try {
      List<RemoteReplicaInfo.ReplicaTokenInfo> replicaTokenInfoList =
          mountPathToReplicaTokenInfos.get(pathWithException1);
      diskTokenPersistor.persist(pathWithException1, replicaTokenInfoList);
      fail("should fail due to I/O exception");
    } catch (IOException e) {
      // expected
    } finally {
      assertTrue("Can't make dir writable", mountPathDir1.setWritable(true));
    }
    // mock I/O exception for 2nd mount path but disk is still available (at least one store is up)
    String pathWithException2 = pathItor.next();
    Mockito.when(mockStoreManager.checkLocalPartitionStatus(any(), any()))
        .thenReturn(ServerErrorCode.ReplicaUnavailable);
    File mountPathDir2 = new File(pathWithException2);
    assertTrue("Can't make dir unwritable", mountPathDir2.setWritable(false));
    try {
      List<RemoteReplicaInfo.ReplicaTokenInfo> replicaTokenInfoList =
          mountPathToReplicaTokenInfos.get(pathWithException2);
      diskTokenPersistor.persist(pathWithException2, replicaTokenInfoList);
      fail("should fail due to I/O exception");
    } catch (IOException e) {
      // expected
    } finally {
      assertTrue("Can't make dir writable", mountPathDir2.setWritable(true));
    }
    // now verify the paths-to-skip set only contains 1st mount path
    Set<String> pathsToSkip = diskTokenPersistor.getMountPathsToSkip();
    assertTrue("mountPathsToSkip should only contain 1st mount path",
        pathsToSkip.size() == 1 && pathsToSkip.contains(pathWithException1));
    // verify the bad disk (at 1st mount path) is skipped
    diskTokenPersistor.write(false);
    // retrieve token from all paths
    for (String mountPath : mountPathToReplicaTokenInfos.keySet()) {
      List<RemoteReplicaInfo.ReplicaTokenInfo> retrievedReplicaTokenInfos = diskTokenPersistor.retrieve(mountPath);
      if (mountPath.equals(pathWithException1)) {
        assertTrue("The token info should be empty", retrievedReplicaTokenInfos.isEmpty());
      } else {
        List<RemoteReplicaInfo.ReplicaTokenInfo> replicaTokenInfoList = mountPathToReplicaTokenInfos.get(mountPath);
        assertEquals("Number of tokens doesn't match.", replicaTokenInfoList.size(), retrievedReplicaTokenInfos.size());
        for (int i = 0; i < replicaTokenInfoList.size(); i++) {
          assertArrayEquals("Token is not correct.", replicaTokenInfoList.get(i).getReplicaToken().toBytes(),
              retrievedReplicaTokenInfos.get(i).getReplicaToken().toBytes());
        }
      }
    }
  }
Test that replica token persistor is able to skip the bad disk if all stores on that disk are down.
@throws Exception
ambry-replication/src/test/java/com/github/ambry/replication/DiskTokenPersistorTest.java