private void verifyDiskFailureSuccess(StorageManager storageManager, StorageManager.DiskFailureHandler handler,
      HelixAdmin helixAdmin, ClusterMap clusterMap, String clusterName, DataNodeId localNode,
      Map<DiskId, List<ReplicaId>> diskToReplicas, DiskId diskToFail) throws Exception {
    long failureCountBefore = storageManager.getStoreMainMetrics().handleDiskFailureCount.getCount();
    long successCountBefore = storageManager.getStoreMainMetrics().handleDiskFailureSuccessCount.getCount();
    List<? extends ReplicaId> allReplicaIds = clusterMap.getReplicaIds(localNode);
    int numDisksInMemory = storageManager.getDiskToDiskManager().size();
    String instanceName = ClusterMapUtils.getInstanceName(localNode.getHostname(), localNode.getPort());
    int failedDisksBefore = handler.getFailedDisks().size();
    int offlineReplicasBefore = getNumberOfReplicaInStateFromMetric("offline", metricRegistry);

    handler.run();
    assertEquals(failureCountBefore + 1, storageManager.getStoreMainMetrics().handleDiskFailureCount.getCount());
    assertEquals(successCountBefore + 1, storageManager.getStoreMainMetrics().handleDiskFailureSuccessCount.getCount());

    assertEquals("1 more disk failed", failedDisksBefore + 1, handler.getFailedDisks().size());
    assertFalse("Cluster should not in maintenance mode", helixAdmin.isInMaintenanceMode(clusterName));
    Thread.sleep(500); // wait for clustermap to sync with zookeeper
    // replicas should be removed from helix clustermap
    List<? extends ReplicaId> healthyReplicaIds = clusterMap.getReplicaIds(localNode);
    Set<? extends ReplicaId> removedReplicas = new HashSet<>(allReplicaIds);
    removedReplicas.removeAll(healthyReplicaIds);
    List<ReplicaId> replicasOnFailedDisk = diskToReplicas.get(diskToFail);
    assertEquals("Replicas should be removed from helix clustermap", new HashSet<>(replicasOnFailedDisk),
        removedReplicas);
    assertEquals("Disk should be unavailable now", HardwareState.UNAVAILABLE, diskToFail.getState());
    assertEquals(
        "Replicas should be reset to offline, number of replica on failed disk is " + replicasOnFailedDisk.size()
            + " offline replica before is " + offlineReplicasBefore,
        offlineReplicasBefore + replicasOnFailedDisk.size(),
        getNumberOfReplicaInStateFromMetric("offline", metricRegistry));
    diskToReplicas.remove(diskToFail);
    int healthyDiskCapacity =
        (int) (diskToReplicas.keySet().stream().mapToLong(DiskId::getRawCapacityInBytes).sum() / 1024 / 1024 / 1024
            * 0.95);
    assertEquals("Disk capacity should be updated", healthyDiskCapacity,
        helixAdmin.getInstanceConfig(clusterName, instanceName)
            .getInstanceCapacityMap()
            .get(HelixParticipant.DISK_KEY)
            .intValue());
    for (ReplicaId replicaId : replicasOnFailedDisk) {
      assertNull(storageManager.getStore(replicaId.getPartitionId(), false));
      assertNull(storageManager.getReplica(replicaId.getPartitionId().toPathString()));
      assertNull(storageManager.getDiskManager(replicaId.getPartitionId()));
    }
    assertEquals(numDisksInMemory - 1, storageManager.getDiskToDiskManager().size());
  }
Verify the state after handling a disk failure.
@param storageManager The {@link StorageManager}.
@param handler The {@link com.github.ambry.store.StorageManager.DiskFailureHandler}.
@param helixAdmin The {@link HelixAdmin}.
@param clusterMap The {@link ClusterMap}.
@param clusterName The cluster name
@param localNode The local node
@param diskToReplicas The map from disk id to a list of replicas
@param diskToFail The disk to fail
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/StorageManagerTest.java