@Test
  public void replicaFromOfflineToBootstrapFailureTest() throws Exception {
    generateConfigs(true, false);
    MockDataNodeId localNode = clusterMap.getDataNodes().get(0);
    List<PartitionId> partitionIds = clusterMap.getAllPartitionIds(null);
    List<ReplicaId> localReplicas = clusterMap.getReplicaIds(localNode);
    MockClusterParticipant mockHelixParticipant = new MockClusterParticipant();
    StorageManager storageManager =
        createStorageManager(localNode, metricRegistry, Collections.singletonList(mockHelixParticipant));
    storageManager.start();
    // 1. get listeners from Helix participant and verify there is a storageManager listener.
    Map<StateModelListenerType, PartitionStateChangeListener> listeners =
        mockHelixParticipant.getPartitionStateChangeListeners();
    assertTrue("Should contain storage manager listener",
        listeners.containsKey(StateModelListenerType.StorageManagerListener));
    // 2. if new bootstrap replica is not found, there should be an exception
    try {
      mockHelixParticipant.onPartitionBecomeBootstrapFromOffline(String.valueOf(partitionIds.size() + 1));
      fail("should fail due to bootstrap replica not found");
    } catch (StateTransitionException e) {
      assertEquals("Error code doesn't match", ReplicaNotFound, e.getErrorCode());
    }

    // 3. test regular store didn't start up (which triggers StoreNotStarted exception)
    ReplicaId replicaId = localReplicas.get(0);
    Store localStore = storageManager.getStore(replicaId.getPartitionId(), true);
    localStore.shutdown();
    try {
      mockHelixParticipant.onPartitionBecomeBootstrapFromOffline(replicaId.getPartitionId().toPathString());
      fail("should fail due to store not started");
    } catch (StateTransitionException e) {
      assertEquals("Error code doesn't match", StoreNotStarted, e.getErrorCode());
    }
    localStore.start();

    // 4. test both failure and success cases regarding new replica addition
    PartitionId newPartition = clusterMap.createNewPartition(Collections.singletonList(localNode));
    assertNull("There should not be any store associated with new partition",
        storageManager.getStore(newPartition, true));
    // find an existing replica that shares disk with new replica
    ReplicaId newReplica = newPartition.getReplicaIds().get(0);
    ReplicaId replicaOnSameDisk =
        localReplicas.stream().filter(r -> r.getDiskId().equals(newReplica.getDiskId())).findFirst().get();
    // test add new store failure by shutting down target diskManager
    storageManager.getDiskManager(replicaOnSameDisk.getPartitionId()).shutdown();
    try {
      mockHelixParticipant.onPartitionBecomeBootstrapFromOffline(newPartition.toPathString());
      fail("should fail due to disk is down");
    } catch (StateTransitionException e) {
      assertEquals("Error code doesn't match", ReplicaOperationFailure, e.getErrorCode());
    }

    shutdownAndAssertStoresInaccessible(storageManager, localReplicas);
  }
test that both success and failure in storage manager when replica becomes BOOTSTRAP from OFFLINE (update
InstanceConfig in Helix is turned off in this test)
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/StorageManagerTest.java