protected void lifeVersionLocalGreaterThanRemote_Delete(boolean localTtlUpdated, boolean remoteTtlUpdated)
      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();

    List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(null);
    for (int i = 0; i < partitionIds.size(); i++) {
      PartitionId partitionId = partitionIds.get(i);
      // add 1 messages to remote host with lifeVersion being 0 and add it local host with lifeVersion being 1.
      StoreKey toDeleteId = addPutMessagesToReplicasOfPartition(partitionId, Arrays.asList(remoteHost), 1).get(0);
      if (remoteTtlUpdated) {
        addTtlUpdateMessagesToReplicasOfPartition(partitionId, toDeleteId, Arrays.asList(remoteHost),
            UPDATED_EXPIRY_TIME_MS);
      }
      addDeleteMessagesToReplicasOfPartition(partitionId, toDeleteId, Collections.singletonList(remoteHost));

      BlobId blobId = (BlobId) toDeleteId;
      short accountId = blobId.getAccountId();
      short containerId = blobId.getContainerId();
      short lifeVersion = 1;
      // first put message has encryption turned on
      boolean toEncrypt = true;

      // create a put message with lifeVersion bigger than 0
      PutMsgInfoAndBuffer msgInfoAndBuffer =
          createPutMessage(toDeleteId, accountId, containerId, toEncrypt, lifeVersion);
      localHost.addMessage(partitionId,
          new MessageInfo(toDeleteId, msgInfoAndBuffer.byteBuffer.remaining(), false, false, false, Utils.Infinite_Time,
              null, accountId, containerId, msgInfoAndBuffer.messageInfo.getOperationTimeMs(), lifeVersion),
          msgInfoAndBuffer.byteBuffer);
      if (localTtlUpdated) {
        addTtlUpdateMessagesToReplicasOfPartition(partitionId, toDeleteId, Collections.singletonList(localHost),
            EXPIRY_TIME_MS, lifeVersion);
      }

      // ensure that the first key is not deleted in the local host
      assertNull(toDeleteId + " should not be deleted in the local host",
          getMessageInfo(toDeleteId, localHost.infosByPartition.get(partitionId), true, false, false));
    }

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

    // Do the replica metadata exchange.
    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++) {
      // we don't have any missing key here.
      assertEquals(0, response.get(i).missingStoreMessages.size());
      remoteReplicaInfos.get(i).setToken(response.get(i).remoteToken);
      PartitionId partitionId = partitionIds.get(i);
      StoreKey key = localHost.infosByPartition.get(partitionId).get(0).getStoreKey();
      assertNull(key + " should not be deleted in the local host",
          getMessageInfo(key, localHost.infosByPartition.get(partitionId), true, false, false));
      if (!localTtlUpdated) {
        assertNull(key + " should not be ttlUpdated in the local host",
            getMessageInfo(key, localHost.infosByPartition.get(partitionId), false, false, true));
      }
    }
  }
Helper function to test when the local lifeVersion is greater than the remote lifeVersion.
@param localTtlUpdated
@param remoteTtlUpdated
@throws Exception
ambry-replication/src/test/java/com/github/ambry/replication/ReplicationTestHelper.java