@Test
  public void replicaThreadLeaderBasedReplicationForTTLUpdatesDeleteAndUndeleteMessagesTest() throws Exception {
    int batchSize = 10;
    Map<DataNodeId, MockHost> hosts = new HashMap<>();
    hosts.put(remoteNodeInLocalDC, remoteHostInLocalDC);
    hosts.put(remoteNodeInRemoteDC, remoteHostInRemoteDC);
    MockNetworkClientFactory mockNetworkClientFactory = new MockNetworkClientFactory(hosts, clusterMap, batchSize,
        new MockFindTokenHelper(new BlobIdFactory(clusterMap), replicationConfig));

    Pair<StorageManager, ReplicationManager> managers =
        createStorageManagerAndReplicationManager(clusterMap, clusterMapConfig, mockHelixParticipant,
            mockNetworkClientFactory);
    StorageManager storageManager = managers.getFirst();
    MockReplicationManager replicationManager = (MockReplicationManager) managers.getSecond();

    // set mock local stores on all remoteReplicaInfos which will used during replication.
    for (PartitionId partitionId : replicationManager.partitionToPartitionInfo.keySet()) {
      localHost.addStore(partitionId, null);
      Store localStore = localHost.getStore(partitionId);
      localStore.start();
      List<RemoteReplicaInfo> remoteReplicaInfos =
          replicationManager.partitionToPartitionInfo.get(partitionId).getRemoteReplicaInfos();
      remoteReplicaInfos.forEach(remoteReplicaInfo -> remoteReplicaInfo.setLocalStore(localStore));
    }

    // get remote replicas and replica thread for remote host on local datacenter
    ReplicaThread intraColoReplicaThread = replicationManager.dataNodeIdToReplicaThread.get(remoteNodeInLocalDC);
    List<RemoteReplicaInfo> remoteReplicaInfosForLocalDC =
        intraColoReplicaThread.getRemoteReplicaInfos().get(remoteNodeInLocalDC);

    // get remote replicas and replica thread for remote host on remote datacenter
    ReplicaThread crossColoReplicaThread = replicationManager.dataNodeIdToReplicaThread.get(remoteNodeInRemoteDC);
    List<RemoteReplicaInfo> remoteReplicaInfosForRemoteDC =
        crossColoReplicaThread.getRemoteReplicaInfos().get(remoteNodeInRemoteDC);

    // mock helix transition state from standby to leader for local leader partitions
    List<? extends ReplicaId> replicaIds = clusterMap.getReplicaIds(replicationManager.dataNodeId);
    for (ReplicaId replicaId : replicaIds) {
      MockReplicaId mockReplicaId = (MockReplicaId) replicaId;
      if (mockReplicaId.getReplicaState() == ReplicaState.LEADER) {
        MockPartitionId mockPartitionId = (MockPartitionId) replicaId.getPartitionId();
        mockHelixParticipant.onPartitionBecomeLeaderFromStandby(mockPartitionId.toPathString());
      }
    }

    List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(null);
    Map<PartitionId, List<StoreKey>> idsToBeIgnoredByPartition = new HashMap<>();
    Map<PartitionId, List<StoreKey>> idsToBeTtlUpdatedByPartition = new HashMap<>();

    /*
    Scenario:
    we have 3 nodes that have replicas belonging to same partitions:
      a) local node
      b) remote node in local DC
      c) remote node in remote DC

    We will have following blob entries in remoteNodeInLocalDC and remoteNodeInRemoteDC:
     1. b0 (PUT)
     2. b1 (PUT) + b1 (TTL_UPDATE)
     3. b2 (PUT) + b2 (TTL_UPDATE) + b2 (DELETE)
     4. b3 (PUT) + b3 (TTL_UPDATE) + b3 (DELETE) + b3 (UN_DELETE)

     Messages order in remoteNodeInLocalDC:
     b0, b1(PUT), b2(PUT), b3(PUT), b1(TTL_UPDATE), b2(TTL_UPDATE), b3(TTL_UPDATE), b2(DELETE),  b3(DELETE), b3(UN_DELETE).

     Messages order in remoteNodeInRemoteDC:
     b0, b1(PUT), b2(PUT), b3(PUT), b3(TTL_UPDATE), b2(TTL_UPDATE), b1(TTL_UPDATE), b3(DELETE), b2(DELETE), b3(UN_DELETE).

     Set batch size to be fetched during each replication cycle = 10 so that all messages will be fetched in one replication cycle.

     Steps:
     1. Exchange metadata with remoteNodeInRemoteDC to fetch all the messages (set the batch_size in connection to large value).
     2. Verify that GET requests will be only be sent for leader replicas while standby replicas will store the metadata exchange information.
     3. Exchange metadata with remoteNodeInLocalDC to fetch all the messages (set the batch_size in connection to large value).
     4. Verify that GET requests will be sent for all replicas as it is intra-dc replication.
     5. After the intra-dc replication, verify that remote token would move forward for all replicas in remoteNodeInRemoteDC.
     6. Verify that message information and buffers are in sync at local host and remote hosts in local and remote data centers.
     */

    for (PartitionId id : partitionIds) {
      List<StoreKey> toBeIgnored = new ArrayList<>();
      List<StoreKey> toBeUndeleted = new ArrayList<>();

      // Adding 4 PUT messages b0, b1, b2, b3 to remoteNodeInLocalDC and remoteNodeInRemoteDC
      List<StoreKey> ids =
          addPutMessagesToReplicasOfPartition(id, Arrays.asList(remoteHostInLocalDC, remoteHostInRemoteDC), 4);

      // ttl update to be added to b1, b2, b3
      List<StoreKey> toBeTtlUpdated = new ArrayList<>(ids);
      toBeTtlUpdated.remove(ids.get(0));

      // delete to be added to b2, b3
      toBeIgnored.add(ids.get(2));
      toBeIgnored.add(ids.get(3));

      // un-delete to be added to b3
      toBeUndeleted.add(ids.get(3));
      toBeIgnored.remove(ids.get(3));

      // Add TTLUpdate records for blobs b1,b2,b3 in remoteNodeInLocalDC and remoteNodeInRemoteDC
      for (int j = 0; j < toBeTtlUpdated.size(); j++) {
        addTtlUpdateMessagesToReplicasOfPartition(id, toBeTtlUpdated.get(j),
            Collections.singletonList(remoteHostInLocalDC), UPDATED_EXPIRY_TIME_MS);
        addTtlUpdateMessagesToReplicasOfPartition(id, toBeTtlUpdated.get(toBeTtlUpdated.size() - 1 - j),
            Collections.singletonList(remoteHostInRemoteDC), UPDATED_EXPIRY_TIME_MS);
      }

      // Add delete records for blobs b2,b3 in remoteNodeInLocalDC and remoteNodeInRemoteDC
      for (int j = 0; j < toBeIgnored.size(); j++) {
        addDeleteMessagesToReplicasOfPartition(id, toBeIgnored.get(j), Collections.singletonList(remoteHostInLocalDC),
            (short) 0, EXPIRY_TIME_MS);
        addDeleteMessagesToReplicasOfPartition(id, toBeIgnored.get(toBeIgnored.size() - 1 - j),
            Collections.singletonList(remoteHostInRemoteDC), (short) 0, EXPIRY_TIME_MS);
      }

      // Add un-delete records for blob b3 with life_version as 1 in remoteNodeInLocalDC and remoteNodeInRemoteDC
      for (StoreKey storeKey : toBeUndeleted) {
        addUndeleteMessagesToReplicasOfPartition(id, storeKey, Arrays.asList(remoteHostInLocalDC, remoteHostInRemoteDC),
            (short) 1);
      }

      // will be used later while comparing the final message records in local and remote nodes
      idsToBeIgnoredByPartition.put(id, toBeIgnored);
      idsToBeTtlUpdatedByPartition.put(id, toBeTtlUpdated);
    }

    // Inter-dc replication
    // Send metadata request to remoteNodeInRemoteDC to fetch missing keys information.
    crossColoReplicaThread.replicate();
    List<ReplicaThread.ExchangeMetadataResponse> responseForRemoteNodeInRemoteDC =
        crossColoReplicaThread.getExchangeMetadataResponsesInEachCycle().get(remoteNodeInRemoteDC);
    assertEquals("Response should contain a response for each replica", remoteReplicaInfosForRemoteDC.size(),
        responseForRemoteNodeInRemoteDC.size());

    // Filter leader replicas to fetch missing keys
    List<RemoteReplicaInfo> leaderReplicas = new ArrayList<>();
    List<ReplicaThread.ExchangeMetadataResponse> exchangeMetadataResponseListForLeaderReplicas = new ArrayList<>();
    crossColoReplicaThread.getLeaderReplicaList(remoteReplicaInfosForRemoteDC, responseForRemoteNodeInRemoteDC,
        leaderReplicas, exchangeMetadataResponseListForLeaderReplicas);

    // verify that only leader partitions in local and remote nodes are chosen for fetching missing messages.
    Set<ReplicaId> remoteLeaderReplicasWithLeaderPartitionsOnLocalNode =
        getRemoteLeaderReplicasWithLeaderPartitionsOnLocalNode(clusterMap, replicationManager.dataNodeId,
            remoteNodeInRemoteDC);
    Set<ReplicaId> leaderReplicaSetInReplicaThread =
        leaderReplicas.stream().map(RemoteReplicaInfo::getReplicaId).collect(Collectors.toSet());
    assertThat("mismatch in leader remote replicas to fetch missing keys",
        remoteLeaderReplicasWithLeaderPartitionsOnLocalNode, is(leaderReplicaSetInReplicaThread));

    // Verify that the remote token will move forward only for leader replicas
    // For standby replicas, token index will remain 0 and metadata information would be stored.
    for (int i = 0; i < remoteReplicaInfosForRemoteDC.size(); i++) {
      if (remoteLeaderReplicasWithLeaderPartitionsOnLocalNode.contains(
          remoteReplicaInfosForRemoteDC.get(i).getReplicaId())) {
        assertEquals("remote Token should be updated for leader replica",
            remoteReplicaInfosForRemoteDC.get(i).getToken(), (responseForRemoteNodeInRemoteDC.get(i).remoteToken));
      } else {
        assertThat("remote Token should not be updated for standby replica",
            remoteReplicaInfosForRemoteDC.get(i).getToken(), not(responseForRemoteNodeInRemoteDC.get(i).remoteToken));
        assertEquals("missing messages in metadata exchange should be stored for standby replica",
            remoteReplicaInfosForRemoteDC.get(i).getExchangeMetadataResponse().missingStoreMessages.size(),
            responseForRemoteNodeInRemoteDC.get(i).missingStoreMessages.size());
      }
    }

    // Intra-dc replication
    intraColoReplicaThread.replicate();
    List<ReplicaThread.ExchangeMetadataResponse> responseForRemoteNodeInLocalDC =
        intraColoReplicaThread.getExchangeMetadataResponsesInEachCycle().get(remoteNodeInLocalDC);

    // Verify that the remote token for all intra-colo replicas has been moved
    for (int i = 0; i < responseForRemoteNodeInLocalDC.size(); i++) {
      assertEquals(remoteReplicaInfosForLocalDC.get(i).getToken(), responseForRemoteNodeInLocalDC.get(i).remoteToken);
    }

    // process missing keys in previous metadata exchange for cross-colo replicas (standby) as they must have
    // arrived via intra-dc replication
    for (RemoteReplicaInfo remoteReplicaInfo : remoteReplicaInfosForRemoteDC) {
      crossColoReplicaThread.processMissingKeysFromPreviousMetadataResponse(remoteReplicaInfo);
    }

    // Verify that the remote token for all inter-colo replicas (standby) has been moved as they must have
    // arrived via intra-dc replication
    for (int i = 0; i < responseForRemoteNodeInLocalDC.size(); i++) {
      assertEquals(remoteReplicaInfosForRemoteDC.get(i).getToken(), responseForRemoteNodeInRemoteDC.get(i).remoteToken);
    }

    // compare the messages and buffers are in sync at local host and remote host1
    checkBlobMessagesAreEqualInLocalAndRemoteHosts(localHost, remoteHostInLocalDC, idsToBeIgnoredByPartition,
        idsToBeTtlUpdatedByPartition);
    // compare the messages and buffers are in sync at local host and remote host1
    checkBlobMessagesAreEqualInLocalAndRemoteHosts(localHost, remoteHostInRemoteDC, idsToBeIgnoredByPartition,
        idsToBeTtlUpdatedByPartition);

    storageManager.shutdown();
  }
Test leader based replication to ensure token is advanced correctly and blob properties ttl
_update, delete, undelete are applied correctly when missing messages in standby replicas
are fetched via intra-dc replication.
@throws Exception
ambry-replication/src/test/java/com/github/ambry/replication/LeaderBasedReplicationTest.java