@Test
  public void replicaThreadTest() throws Exception {
    MockClusterMap clusterMap = new MockClusterMap();
    Pair<MockHost, MockHost> localAndRemoteHosts = getLocalAndRemoteHosts(clusterMap);
    MockHost localHost = localAndRemoteHosts.getFirst();
    MockHost remoteHost = localAndRemoteHosts.getSecond();
    List<MockHost> remoteHosts = new ArrayList<>();
    remoteHosts.add(remoteHost);
    // Add one more remote host
    PartitionId specialPartitionId = clusterMap.getWritablePartitionIds(MockClusterMap.SPECIAL_PARTITION_CLASS).get(0);
    MockHost remoteHost2 = new MockHost(specialPartitionId.getReplicaIds().get(2).getDataNodeId(), clusterMap);
    remoteHosts.add(remoteHost2);

    MockStoreKeyConverterFactory storeKeyConverterFactory = new MockStoreKeyConverterFactory(null, null);
    storeKeyConverterFactory.setConversionMap(new HashMap<>());
    storeKeyConverterFactory.setReturnInputIfAbsent(true);
    MockStoreKeyConverterFactory.MockStoreKeyConverter storeKeyConverter =
        storeKeyConverterFactory.getStoreKeyConverter();

    short blobIdVersion = CommonTestUtils.getCurrentBlobIdVersion();
    List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(null);
    Map<PartitionId, List<StoreKey>> idsToBeIgnoredByPartition = new HashMap<>();
    List<MockHost> allHosts = new ArrayList<>();
    allHosts.add(localHost);
    allHosts.addAll(remoteHosts);
    for (int i = 0; i < partitionIds.size(); i++) {
      List<StoreKey> idsToBeIgnored = new ArrayList<>();
      PartitionId partitionId = partitionIds.get(i);
      // add 6 messages to both hosts.
      StoreKey toDeleteId = addPutMessagesToReplicasOfPartition(partitionId, allHosts, 6).get(0);

      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);
      for (MockHost rHost : remoteHosts) {
        rHost.addMessage(partitionId,
            new MessageInfo(id, msgInfoAndBuffer.byteBuffer.remaining(), 1, accountId, containerId,
                msgInfoAndBuffer.messageInfo.getOperationTimeMs()), msgInfoAndBuffer.byteBuffer);
      }
      idsToBeIgnored.add(id);

      // add 3 messages to the remote host only
      addPutMessagesToReplicasOfPartition(partitionId, remoteHosts, 3);

      accountId = Utils.getRandomShort(TestUtils.RANDOM);
      containerId = Utils.getRandomShort(TestUtils.RANDOM);
      toEncrypt = TestUtils.RANDOM.nextBoolean();
      // add a corrupt message to the remote host only
      id = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, ClusterMap.UNKNOWN_DATACENTER_ID, accountId, containerId,
          partitionId, toEncrypt, BlobId.BlobDataType.DATACHUNK);
      msgInfoAndBuffer = createPutMessage(id, accountId, containerId, toEncrypt);
      byte[] data = msgInfoAndBuffer.byteBuffer.array();
      // flip every bit in the array
      for (int j = 0; j < data.length; j++) {
        data[j] ^= 0xFF;
      }
      for (MockHost rHost : remoteHosts) {
        rHost.addMessage(partitionId, msgInfoAndBuffer.messageInfo, msgInfoAndBuffer.byteBuffer);
      }
      idsToBeIgnored.add(id);

      // add 3 messages to the remote host only
      addPutMessagesToReplicasOfPartition(partitionId, remoteHosts, 3);

      // add delete record for the very first blob in the remote host only
      addDeleteMessagesToReplicasOfPartition(partitionId, toDeleteId, remoteHosts);
      // PUT and DELETE a blob in the remote host only
      id = addPutMessagesToReplicasOfPartition(partitionId, remoteHosts, 1).get(0);
      addDeleteMessagesToReplicasOfPartition(partitionId, id, remoteHosts);
      idsToBeIgnored.add(id);

      addPutMessagesToReplicasOfPartition(partitionId, remoteHosts, 3);

      idsToBeIgnoredByPartition.put(partitionId, idsToBeIgnored);

      // 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));
    }

    StoreKeyFactory storeKeyFactory = new BlobIdFactory(clusterMap);
    Transformer transformer = new BlobIdTransformer(storeKeyFactory, storeKeyConverter);
    int batchSize = 4;
    Pair<Map<DataNodeId, List<RemoteReplicaInfo>>, ReplicaThread> replicasAndThread =
        getRemoteReplicasAndReplicaThread(batchSize, clusterMap, localHost, storeKeyConverter, transformer, null, null,
            remoteHosts.toArray(new MockHost[0]));
    Map<DataNodeId, List<RemoteReplicaInfo>> replicasToReplicate = replicasAndThread.getFirst();
    ReplicaThread replicaThread = replicasAndThread.getSecond();

    for (MockHost rHost : remoteHosts) {
      Map<PartitionId, List<ByteBuffer>> missingBuffers = rHost.getMissingBuffers(localHost.buffersByPartition);
      for (Map.Entry<PartitionId, List<ByteBuffer>> entry : missingBuffers.entrySet()) {
        assertEquals("Missing buffers count mismatch", 14, entry.getValue().size());
      }
    }

    int[] missingKeysCounts = {0, 1, 4, 2, 3};
    int[] missingBuffersCount = {13, 12, 9, 7, 4};
    int expectedIndex = 0;
    int missingBuffersIndex = 0;

    for (int missingKeysCount : missingKeysCounts) {
      expectedIndex = assertMissingKeysAndFixMissingStoreKeys(expectedIndex, batchSize, missingKeysCount, replicaThread,
          replicasToReplicate);

      Map<PartitionId, List<ByteBuffer>> missingBuffers = remoteHost.getMissingBuffers(localHost.buffersByPartition);
      for (Map.Entry<PartitionId, List<ByteBuffer>> entry : missingBuffers.entrySet()) {
        assertEquals("Missing buffers count mismatch for iteration count " + missingBuffersIndex,
            missingBuffersCount[missingBuffersIndex], entry.getValue().size());
      }
      missingBuffersIndex++;
    }

    // All partitions are sync with remote host
    replicaThread.replicate();
    for (MockHost rHost : remoteHosts) {
      List<ReplicaThread.ExchangeMetadataResponse> response =
          replicaThread.getExchangeMetadataResponsesInEachCycle().get(rHost.dataNodeId);
      List<RemoteReplicaInfo> remoteReplicaInfos = replicasToReplicate.get(rHost.dataNodeId);
      assertEquals("Response should contain a response for each replica", remoteReplicaInfos.size(), response.size());
      for (int i = 0; i < response.size(); i++) {
        assertEquals(0, response.get(i).missingStoreMessages.size());
        assertEquals(expectedIndex, ((MockFindToken) response.get(i).remoteToken).getIndex());
        assertEquals("Token should have been set correctly", response.get(i).remoteToken,
            replicasToReplicate.get(rHost.dataNodeId).get(i).getToken());
      }
    }
    time.sleep(10 * 1000);
    replicaThread.replicate();
    for (MockHost rHost : remoteHosts) {
      // 1 expired + 1 corrupt + 1 put (never present) + 1 deleted (never present) expected missing buffers
      verifyNoMoreMissingKeysAndExpectedMissingBufferCount(rHost, localHost, replicaThread, replicasToReplicate,
          idsToBeIgnoredByPartition, storeKeyConverter, expectedIndex, 4);
    }
  }
Tests {@link ReplicaThread#exchangeMetadata(ConnectedChannel, List)} and
{@link ReplicaThread#fixMissingStoreKeys(ConnectedChannel, List, List, boolean)} for valid puts, deletes, expired keys and
Tests {@link ReplicaThread#replicate()} for valid puts, deletes, expired keys and
corrupt blobs.
@throws Exception
ambry-replication/src/test/java/com/github/ambry/replication/ReplicationTest.java