@Test
  public void deletionAfterMetadataExchangeTest() throws Exception {
    int batchSize = 400;
    ReplicationTestSetup testSetup = new ReplicationTestSetup(batchSize);
    short blobIdVersion = CommonTestUtils.getCurrentBlobIdVersion();
    List<PartitionId> partitionIds = testSetup.partitionIds;
    MockHost remoteHost = testSetup.remoteHost;
    MockHost localHost = testSetup.localHost;
    Map<PartitionId, Set<StoreKey>> idsToExpectByPartition = new HashMap<>();
    for (int i = 0; i < partitionIds.size(); i++) {
      PartitionId partitionId = partitionIds.get(i);

      // add 5 messages to remote host only.
      Set<StoreKey> expectedIds =
          new HashSet<>(addPutMessagesToReplicasOfPartition(partitionId, Collections.singletonList(remoteHost), 5));

      short accountId = Utils.getRandomShort(TestUtils.RANDOM);
      short containerId = Utils.getRandomShort(TestUtils.RANDOM);
      boolean toEncrypt = TestUtils.RANDOM.nextBoolean();

      // add an expired message to the remote host only
      StoreKey id =
          new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, ClusterMap.UNKNOWN_DATACENTER_ID, accountId, containerId,
              partitionId, toEncrypt, BlobId.BlobDataType.DATACHUNK);
      PutMsgInfoAndBuffer msgInfoAndBuffer = createPutMessage(id, accountId, containerId, toEncrypt);
      remoteHost.addMessage(partitionId,
          new MessageInfo(id, msgInfoAndBuffer.byteBuffer.remaining(), 1, accountId, containerId,
              msgInfoAndBuffer.messageInfo.getOperationTimeMs()), msgInfoAndBuffer.byteBuffer);

      // add 3 messages to the remote host only
      expectedIds.addAll(addPutMessagesToReplicasOfPartition(partitionId, Collections.singletonList(remoteHost), 3));

      // delete the very first blob in the remote host only (and delete it from expected list)
      Iterator<StoreKey> iter = expectedIds.iterator();
      addDeleteMessagesToReplicasOfPartition(partitionId, iter.next(), Collections.singletonList(remoteHost));
      iter.remove();

      // PUT and DELETE a blob in the remote host only
      id = addPutMessagesToReplicasOfPartition(partitionId, Collections.singletonList(remoteHost), 1).get(0);
      addDeleteMessagesToReplicasOfPartition(partitionId, id, Collections.singletonList(remoteHost));

      idsToExpectByPartition.put(partitionId, expectedIds);
    }

    Map<PartitionId, StoreKey> deletedStoreKeys = new HashMap<>();
    testSetup.replicaThread.setExchangeMetadataListener((partitionId, exception) -> {
      // Now delete a message in the remote before doing the Get requests (for every partition). Remove these keys from
      // expected key set. Even though they are requested, they should not go into the local store. However, this cycle
      // of replication must be successful.
      Iterator<StoreKey> iter = idsToExpectByPartition.get(partitionId).iterator();
      StoreKey keyToDelete = iter.next();
      try {
        addDeleteMessagesToReplicasOfPartition(partitionId, keyToDelete, Collections.singletonList(remoteHost));
      } catch (Exception e) {
      }
      iter.remove();
      deletedStoreKeys.put(partitionId, keyToDelete);
    });

    testSetup.replicaThread.replicate();
    Set<StoreKey> storeKeysToExpect =
        idsToExpectByPartition.values().stream().flatMap(Collection::stream).collect(Collectors.toSet());
    Set<StoreKey> missingStoreKeys = testSetup.replicaThread.getExchangeMetadataResponsesInEachCycle()
        .get(remoteHost.dataNodeId)
        .stream()
        .map(ReplicaThread.ExchangeMetadataResponse::getMissingStoreKeys)
        .flatMap(Collection::stream)
        .collect(Collectors.toSet());
    Assert.assertEquals(storeKeysToExpect.size() + deletedStoreKeys.size(), missingStoreKeys.size());
    // Missing store keys are the original key set before deletes
    storeKeysToExpect.addAll(deletedStoreKeys.values());
    Assert.assertEquals("Actual keys in Exchange Metadata Response different from expected", storeKeysToExpect,
        missingStoreKeys);

    Assert.assertEquals(idsToExpectByPartition.keySet(), localHost.infosByPartition.keySet());
    Assert.assertEquals("Actual keys in Exchange Metadata Response different from expected",
        idsToExpectByPartition.values().stream().flatMap(Collection::stream).collect(Collectors.toSet()),
        localHost.infosByPartition.values()
            .stream()
            .flatMap(Collection::stream)
            .map(MessageInfo::getStoreKey)
            .collect(Collectors.toSet()));
  }
Test the case where a blob gets deleted after a replication metadata exchange completes and identifies the blob as
a candidate. The subsequent GetRequest should succeed as Replication makes a Include_All call and succeed without
exceptions. The blob should not be put locally.
ambry-replication/src/test/java/com/github/ambry/replication/ReplicationTest.java