@Test
  public void storeRemoveDirectoryAndRestartTestWithDiskFailure() throws Exception {
    MockDataNodeId dataNode = clusterMap.getDataNodes().get(0);
    List<ReplicaId> replicas = clusterMap.getReplicaIds(dataNode);
    List<String> mountPaths = dataNode.getMountPaths();
    String badDiskMountPath = mountPaths.get(RANDOM.nextInt(mountPaths.size()));
    List<ReplicaId> badReplicas =
        replicas.stream().filter(r -> r.getMountPath().equals(badDiskMountPath)).collect(Collectors.toList());

    generateConfigs(true, false, false, 2, true, false);
    StorageManager storageManager = createStorageManager(dataNode, metricRegistry, null);
    storageManager.start();
    storageManager.shutdown();

    // Now corrupt all the replicas on the disk except for the last one
    // So we will remove the bad directories and restart blob stores
    for (ReplicaId replica : badReplicas.subList(0, badReplicas.size() - 1)) {
      corruptFirstLogSegment(replica);
    }
    storageManager = createStorageManager(dataNode, metricRegistry, null);
    storageManager.start();
    // All replicas should have its blob store
    for (ReplicaId replica : badReplicas) {
      assertNotNull(storageManager.getStore(replica.getPartitionId(), false));
    }
    storageManager.shutdown();

    // Now corrupt all the replicas on the disk
    for (ReplicaId replica : badReplicas) {
      corruptFirstLogSegment(replica);
    }
    storageManager = createStorageManager(dataNode, metricRegistry, null);
    storageManager.start();
    // All replicas should not have its blob store
    for (ReplicaId replica : badReplicas) {
      assertNull(storageManager.getStore(replica.getPartitionId(), false));
    }
    storageManager.shutdown();
  }
Test the case where the blob store can't get started and we need to remove the directory and then restart the blobstore.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/StorageManagerTest.java