@Test
  public void emptyReplicaFromInactiveToOfflineTest() throws Exception {
    MockClusterMap clusterMap = new MockClusterMap();
    ClusterMapConfig clusterMapConfig = new ClusterMapConfig(verifiableProperties);
    MockHelixParticipant.metricRegistry = new MetricRegistry();
    MockHelixParticipant mockHelixParticipant = new MockHelixParticipant(clusterMapConfig);
    Pair<StorageManager, ReplicationManager> managers =
        createStorageManagerAndReplicationManager(clusterMap, clusterMapConfig, mockHelixParticipant);
    StorageManager storageManager = managers.getFirst();
    MockReplicationManager replicationManager = (MockReplicationManager) managers.getSecond();
    PartitionId existingPartition = replicationManager.partitionToPartitionInfo.keySet().iterator().next();
    Store localStore = storageManager.getStore(existingPartition);
    // Check size of empty store is 18 bytes (header bytes)
    assertEquals("Mismatch in size of empty store", 18, localStore.getSizeInBytes());
    // Create a new thread and trigger INACTIVE -> OFFLINE transition
    ReplicaId localReplica = storageManager.getReplica(existingPartition.toPathString());
    // put a decommission-in-progress file into local store dir
    File decommissionFile = new File(localReplica.getReplicaPath(), "decommission_in_progress");
    assertTrue("Couldn't create decommission file in local store", decommissionFile.createNewFile());
    decommissionFile.deleteOnExit();
    assertNotSame("Before disconnection, the local store state shouldn't be OFFLINE", ReplicaState.OFFLINE,
        localStore.getCurrentState());
    mockHelixParticipant.registerPartitionStateChangeListener(StateModelListenerType.ReplicationManagerListener,
        replicationManager.replicationListener);
    CountDownLatch participantLatch = new CountDownLatch(1);
    replicationManager.listenerExecutionLatch = new CountDownLatch(1);
    Utils.newThread(() -> {
      mockHelixParticipant.onPartitionBecomeOfflineFromInactive(existingPartition.toPathString());
      participantLatch.countDown();
    }, false).start();
    assertTrue("Partition state change listener in ReplicationManager didn't get called within 1 sec",
        replicationManager.listenerExecutionLatch.await(1, TimeUnit.SECONDS));
    // the state of local store should be updated to OFFLINE
    assertEquals("Local store state is not expected", ReplicaState.OFFLINE, localStore.getCurrentState());
    // Since replica is empty, no sync-up is needed and transition should be able to proceed.
    assertTrue("Inactive-To-Offline transition didn't complete within 1 sec",
        participantLatch.await(1, TimeUnit.SECONDS));
    assertTrue("Local store should be running since it is stopped on offline to dropped transition",
        localStore.isStarted());
    storageManager.shutdown();
  }
Test INACTIVE -> OFFLINE transition for empty replica (Corner case)
ambry-replication/src/test/java/com/github/ambry/replication/ReplicationTest.java