@Test
  public void testClusterManagerReplicaStatusDelegateUse() throws StoreException, IOException, InterruptedException {
    //Setup threshold test properties, replicaId, mock replica status delegate
    // TODO: while the compactor is upgraded to understand TTL updates, this hack ensures that this test (which uses
    // TODO: compaction for segmented logs) never encounters TTL updates
    if (isLogSegmented) {
      cleanup();
      scheduler = Utils.newScheduler(1, false);
      storeStatsScheduler = Utils.newScheduler(1, false);
      setupTestState(false, false);
    }
    properties.setProperty("store.set.local.partition.state.enabled", Boolean.toString(true));
    //Setup threshold test properties, replicaId, mock write status delegate
    StoreConfig defaultConfig = changeThreshold(65, 5, true);
    MockReplicaId replicaId = getMockReplicaId(tempDirStr);
    ReplicaStatusDelegate replicaStatusDelegate = mock(ReplicaStatusDelegate.class);
    when(replicaStatusDelegate.unseal(any())).thenReturn(true);
    when(replicaStatusDelegate.partialSeal(any())).thenReturn(true);
    when(replicaStatusDelegate.seal(any())).thenReturn(true);

    //Restart store
    reloadStore(defaultConfig, replicaId, Collections.singletonList(replicaStatusDelegate));

    //Check that after start, replicaStatusDelegate is called to enable replica if it was previously disabled
    verify(replicaStatusDelegate, times(1)).enableReplica(replicaId);

    //Verify that putting in data that doesn't go over the threshold doesn't trigger the delegate
    put(1, 50, Utils.Infinite_Time);
    verify(replicaStatusDelegate, times(0)).seal(replicaId);

    //Verify that after putting in enough data, the store goes to read only
    // setupTestState already have created 3 log segments, there we create another 4 segments, it should
    // be enough to fill up to 65% of the log capacity.
    List<MockId> addedIds = put(4, (long) (SEGMENT_CAPACITY * 0.8), Utils.Infinite_Time);
    verify(replicaStatusDelegate, times(1)).seal(replicaId);

    //Assumes ClusterParticipant sets replicaId status to true
    replicaId.setSealedState(ReplicaSealStatus.SEALED);

    //Change config threshold but with delegate disabled, verify that nothing happens (store doesn't get unsealed)
    reloadStore(changeThreshold(99, 1, false), replicaId, Collections.singletonList(replicaStatusDelegate));
    verify(replicaStatusDelegate, times(0)).unseal(replicaId);

    //Change config threshold to higher, see that it gets changed to unsealed on reset
    reloadStore(changeThreshold(99, 1, true), replicaId, Collections.singletonList(replicaStatusDelegate));
    if (isLogSegmented) {
      verify(replicaStatusDelegate, times(1)).partialSeal(replicaId);
    } else {
      verify(replicaStatusDelegate, times(2)).partialSeal(replicaId);
    }
    replicaId.setSealedState(ReplicaSealStatus.NOT_SEALED);

    //Reset thresholds, verify that it changed back
    reloadStore(defaultConfig, replicaId, Collections.singletonList(replicaStatusDelegate));
    verify(replicaStatusDelegate, times(2)).seal(replicaId);
    replicaId.setSealedState(ReplicaSealStatus.SEALED);

    //Remaining tests only relevant for segmented logs
    if (isLogSegmented) {
      //Delete added data
      for (MockId addedId : addedIds) {
        delete(addedId);
      }

      //Need to restart blob otherwise compaction will ignore segments in journal (which are all segments right now).
      //By restarting, only last segment will be in journal
      reloadStore(defaultConfig, replicaId, Collections.singletonList(replicaStatusDelegate));
      verify(replicaStatusDelegate, times(4)).enableReplica(replicaId);

      //Advance time by 8 days, call compaction to compact segments with deleted data, then verify
      //that the store is now read-write
      time.sleep(TimeUnit.DAYS.toMillis(8));
      store.compact(store.getCompactionDetails(new CompactAllPolicy(defaultConfig, time)),
          new byte[PUT_RECORD_SIZE * 2 + 1]);
      if (isLogSegmented) {
        verify(replicaStatusDelegate, times(1)).partialSeal(replicaId);
      } else {
        verify(replicaStatusDelegate, times(2)).partialSeal(replicaId);
      }

      //Test if replicaId is erroneously true that it updates the status upon startup
      replicaId.setSealedState(ReplicaSealStatus.SEALED);
      reloadStore(defaultConfig, replicaId, Collections.singletonList(replicaStatusDelegate));
      if(isLogSegmented) {
        verify(replicaStatusDelegate, times(1)).partialSeal(replicaId);
      } else {
        verify(replicaStatusDelegate, times(3)).partialSeal(replicaId);
      }
    }
    store.shutdown();
    properties.setProperty("store.set.local.partition.state.enabled", Boolean.toString(false));
  }
Tests blob store use of {@link ReplicaStatusDelegate}
@throws StoreException
ambry-store/src/test/java/com/github/ambry/store/BlobStoreTest.java