@Test
  public void replicaThreadLifeVersionLocalLessThanRemote_MissingPuts() 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);

    List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(null);
    Map<PartitionId, List<StoreKey>> idsToBeIgnoredByPartition = new HashMap<>();
    Map<PartitionId, List<StoreKey>> idsToBeTtlUpdatedByPartition = new HashMap<>();
    short lifeVersion = 1;
    for (int i = 0; i < partitionIds.size(); i++) {
      List<StoreKey> toBeIgnored = new ArrayList<>();
      List<StoreKey> toBeTtlUpdated = new ArrayList<>();
      PartitionId partitionId = partitionIds.get(i);
      // Adding 1 put to remoteHost at lifeVersion 0
      List<StoreKey> ids = addPutMessagesToReplicasOfPartition(partitionId, Collections.singletonList(remoteHost), 1);
      // Adding 1 put to remoteHost at lifeVersion 1
      ids.addAll(
          addPutMessagesToReplicasOfPartition(partitionId, Collections.singletonList(remoteHost), lifeVersion, 1));

      // Adding one put to remoteHost at lifeVersion 1, which would be ttl updated later at lifeVersion 1
      StoreKey toTtlUpdateId =
          addPutMessagesToReplicasOfPartition(partitionId, Collections.singletonList(remoteHost), lifeVersion, 1).get(
              0);
      ids.add(toTtlUpdateId);
      addTtlUpdateMessagesToReplicasOfPartition(partitionId, toTtlUpdateId, Collections.singletonList(remoteHost),
          UPDATED_EXPIRY_TIME_MS, lifeVersion);
      toBeTtlUpdated.add(toTtlUpdateId);

      // Adding one put to remoteHost at lifeVersion 0, which would be ttl updated later at lifeVersion 1
      toTtlUpdateId = addPutMessagesToReplicasOfPartition(partitionId, Collections.singletonList(remoteHost), 1).get(0);
      ids.add(toTtlUpdateId);
      addTtlUpdateMessagesToReplicasOfPartition(partitionId, toTtlUpdateId, Collections.singletonList(remoteHost),
          UPDATED_EXPIRY_TIME_MS, lifeVersion);
      toBeTtlUpdated.add(toTtlUpdateId);

      // Adding one put to remoteHost, which would be deleted later
      StoreKey toDeleteId =
          addPutMessagesToReplicasOfPartition(partitionId, Arrays.asList(remoteHost), lifeVersion, 1).get(0);
      ids.add(toDeleteId);
      addDeleteMessagesToReplicasOfPartition(partitionId, toDeleteId, Collections.singletonList(remoteHost),
          lifeVersion, EXPIRY_TIME_MS);
      toBeIgnored.add(toDeleteId);

      // Adding one put to remoteHost, which would be ttl updated and deleted later
      StoreKey toDeleteAndTtlUpdateId =
          addPutMessagesToReplicasOfPartition(partitionId, Arrays.asList(remoteHost), lifeVersion, 1).get(0);
      ids.add(toDeleteAndTtlUpdateId);
      addTtlUpdateMessagesToReplicasOfPartition(partitionId, toDeleteAndTtlUpdateId,
          Collections.singletonList(remoteHost), UPDATED_EXPIRY_TIME_MS, lifeVersion);
      toBeTtlUpdated.add(toDeleteAndTtlUpdateId);
      addDeleteMessagesToReplicasOfPartition(partitionId, toDeleteAndTtlUpdateId, Collections.singletonList(remoteHost),
          lifeVersion, UPDATED_EXPIRY_TIME_MS);
      toBeIgnored.add(toDeleteAndTtlUpdateId);

      // Adding one put to remoteHost at lifeVersion 0, delete it and then add undelete at lifeVersion 1
      StoreKey deleteAndUndeleteId =
          addPutMessagesToReplicasOfPartition(partitionId, Collections.singletonList(remoteHost), 1).get(0);
      ids.add(deleteAndUndeleteId);
      addDeleteMessagesToReplicasOfPartition(partitionId, deleteAndUndeleteId, Collections.singletonList(remoteHost),
          (short) 0, EXPIRY_TIME_MS);
      addUndeleteMessagesToReplicasOfPartition(partitionId, deleteAndUndeleteId, Collections.singletonList(remoteHost),
          lifeVersion);

      idsToBeIgnoredByPartition.put(partitionId, toBeIgnored);
      idsToBeTtlUpdatedByPartition.put(partitionId, toBeTtlUpdated);

      // Adding one put to both remote and local host.
      ids.addAll(
          addPutMessagesToReplicasOfPartition(partitionId, Arrays.asList(localHost, remoteHost), lifeVersion, 1));
    }

    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();

    int missingKeyCount = 5;
    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(missingKeyCount, response.get(i).missingStoreMessages.size());
      assertEquals("Token should have been set correctly", response.get(i).remoteToken,
          remoteReplicaInfos.get(i).getToken());
    }
    // Don't compare buffers here, PutBuffer might be different since we might change the lifeVersion.
    for (Map.Entry<PartitionId, List<MessageInfo>> localInfoEntry : localHost.infosByPartition.entrySet()) {
      assertEquals("MessageInfo number mismatch", 8, localInfoEntry.getValue().size());
    }

    checkBlobMessagesAreEqualInLocalAndRemoteHosts(localHost, remoteHost, idsToBeIgnoredByPartition,
        idsToBeTtlUpdatedByPartition);
  }
Tests when the local store missing put records with lifeVersion greater than 0
ambry-replication/src/test/java/com/github/ambry/replication/ReplicationTest.java