@Test
  public void replicaThreadLifeVersionLocalLessThanRemote_FinalState_NotDelete() throws Exception {
    MockClusterMap clusterMap = new MockClusterMap();
    Pair<MockHost, MockHost> localAndRemoteHosts = getLocalAndRemoteHosts(clusterMap);
    MockHost localHost = localAndRemoteHosts.getFirst();
    MockHost remoteHost = localAndRemoteHosts.getSecond();
    MockStoreKeyConverterFactory storeKeyConverterFactory = new MockStoreKeyConverterFactory(null, null);
    storeKeyConverterFactory.setConversionMap(new HashMap<>());
    storeKeyConverterFactory.setReturnInputIfAbsent(true);
    MockStoreKeyConverterFactory.MockStoreKeyConverter storeKeyConverter =
        storeKeyConverterFactory.getStoreKeyConverter();
    Map<StoreKey, StoreKey> conversionMap = new HashMap<>();
    storeKeyConverter.setConversionMap(conversionMap);
    StoreKeyFactory storeKeyFactory = new BlobIdFactory(clusterMap);
    Transformer transformer = new BlobIdTransformer(storeKeyFactory, storeKeyConverter);

    Map<PartitionId, List<StoreKey>> idsByPartition = new HashMap<>();
    Map<PartitionId, StoreKey> idsToBeIgnoredByPartition = new HashMap<>();
    List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(null);
    // Remote host has final state as "Not Delete", it would be Put, Ttl update or Undelete. Put and Undelete are practically
    // the same. So we can create two separate set of remote host, with ttl update or without ttl update.
    //
    // When the remote has ttl update, assuming it's P1, T1, then the local can be
    // 1 Missing -> [P1, T1]
    // 2 P0 -> [U1, T1]
    // 3 P0, T0 -> [U1]
    // 4 P0, T0, D0 -> [U1]
    // 5 P0, D0 -> [U1, T1]
    //
    // When the remote has not ttl update, assuming it's P1, then the local can be
    // 1 Missing -> [P1]
    // 2 P0 -> [U1]
    // 3 P0, T0 -> [U1]
    // 4 P0, T0, D0 -> [U1]
    // 5 P0, D0 -> [U1]
    for (int i = 0; i < partitionIds.size(); i++) {
      PartitionId partitionId = partitionIds.get(i);
      List<StoreKey> ids = addPutMessagesToReplicasOfPartition(partitionId, Arrays.asList(remoteHost), (short) 1, 5);
      for (StoreKey id : ids) {
        addTtlUpdateMessagesToReplicasOfPartition(partitionId, id, Collections.singletonList(remoteHost),
            UPDATED_EXPIRY_TIME_MS, (short) 1);
      }

      // 1 Missing
      StoreKey id = ids.get(0);

      // 2 P0 -> [U1, T1]
      id = ids.get(1);
      addPutMessagesToReplicasOfPartition(Collections.singletonList(id), Collections.singletonList(localHost));

      // 3 P0, T0 -> [U1]
      id = ids.get(2);
      addPutMessagesToReplicasOfPartition(Collections.singletonList(id), Collections.singletonList(localHost));
      addTtlUpdateMessagesToReplicasOfPartition(partitionId, id, Arrays.asList(localHost), UPDATED_EXPIRY_TIME_MS,
          (short) 0);

      // 4 P0, T0, D0 -> [U1]
      id = ids.get(3);
      addPutMessagesToReplicasOfPartition(Collections.singletonList(id), Collections.singletonList(localHost));
      addTtlUpdateMessagesToReplicasOfPartition(partitionId, id, Arrays.asList(localHost), UPDATED_EXPIRY_TIME_MS,
          (short) 0);
      addDeleteMessagesToReplicasOfPartition(partitionId, id, Collections.singletonList(localHost), (short) 0,
          UPDATED_EXPIRY_TIME_MS);

      // 5 P, D -> [U, T, D]
      id = ids.get(4);
      addPutMessagesToReplicasOfPartition(Collections.singletonList(id), Collections.singletonList(localHost));
      addDeleteMessagesToReplicasOfPartition(partitionId, id, Collections.singletonList(localHost), (short) 0,
          UPDATED_EXPIRY_TIME_MS);
    }

    int batchSize = 100;
    Pair<Map<DataNodeId, List<RemoteReplicaInfo>>, ReplicaThread> replicasAndThread =
        getRemoteReplicasAndReplicaThread(batchSize, clusterMap, localHost, storeKeyConverter, transformer, null, null,
            remoteHost);
    List<RemoteReplicaInfo> remoteReplicaInfos = replicasAndThread.getFirst().get(remoteHost.dataNodeId);
    ReplicaThread replicaThread = replicasAndThread.getSecond();

    // There is one missing key
    replicaThread.replicate();
    List<ReplicaThread.ExchangeMetadataResponse> response =
        replicaThread.getExchangeMetadataResponsesInEachCycle().get(remoteHost.dataNodeId);
    assertEquals("Response should contain a response for each replica", remoteReplicaInfos.size(), response.size());
    for (int i = 0; i < response.size(); i++) {
      assertEquals(1, response.get(i).missingStoreMessages.size());
    }

    // Before exchange metadata, the number of message infos in local host is 8. Exchange metadata would add another 8.
    for (Map.Entry<PartitionId, List<MessageInfo>> localInfoEntry : localHost.infosByPartition.entrySet()) {
      assertEquals("MessageInfo number mismatch", 16, localInfoEntry.getValue().size());
    }

    for (Map.Entry<PartitionId, List<StoreKey>> idsEntry : idsByPartition.entrySet()) {
      List<MessageInfo> remoteInfos = remoteHost.infosByPartition.get(idsEntry.getKey());
      List<MessageInfo> localInfos = localHost.infosByPartition.get(idsEntry.getKey());

      for (StoreKey id : idsEntry.getValue()) {
        if (!idsToBeIgnoredByPartition.get(idsEntry.getKey()).equals(id)) {
          MessageInfo localInfo = getMergedMessageInfo(id, localInfos);
          MessageInfo remoteInfo = getMergedMessageInfo(id, remoteInfos);
          assertTrue(localInfo.isDeleted());
          assertTrue(remoteInfo.isDeleted());
          assertTrue(localInfo.isTtlUpdated());
          assertTrue(remoteInfo.isTtlUpdated());
          assertEquals(localInfo.getLifeVersion(), remoteInfo.getLifeVersion());
        }
      }
    }
  }
Tests when lifeVersion in local is less than the lifeVersion in remote and the final state is not
delete, it would be Put, TtlUpdate or Undelete.
@throws Exception
ambry-replication/src/test/java/com/github/ambry/replication/ReplicationTest.java