@Test
  public void testSelectReplicas() throws ReflectiveOperationException {
    // Give hosts a name
    AtomicInteger ai = new AtomicInteger(0);
    int Z = 'Z';
    clustermap.getDataNodes().forEach(d -> d.setHostname(String.valueOf((char)(Z - (ai.getAndIncrement() % 26)))));

    // Create a test-thread
    VcrReplicaThread rthread =
        new VcrReplicaThread("vcrReplicaThreadTest", new MockFindTokenHelper(), clustermap,
            new AtomicInteger(0), clustermap.getDataNodes().get(0), null, null,
            null, null, false,
            clustermap.getDataNodes().get(0).getDatacenterName(), null, null,
            null, null, null, null,
            properties);

    // Assign replicas to test-thread
    List<PartitionId> partitions = clustermap.getAllPartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS);
    Map<DataNodeId, List<RemoteReplicaInfo>> nodes = new HashMap<>();
    partitions.forEach(partition -> partition.getReplicaIds().forEach(replica -> {
      RemoteReplicaInfo rinfo =
          new RemoteReplicaInfo(replica, null, null, null, 0,
              SystemTime.getInstance(), null);
      rthread.addRemoteReplicaInfo(rinfo);
      // Group by datanode
      DataNodeId dnode = replica.getDataNodeId();
      List rlist = nodes.getOrDefault(dnode, new ArrayList<>());
      rlist.add(rinfo);
      nodes.putIfAbsent(dnode, rlist);
    }));

    // Call custom-filter. Each time its called, it picks one replica per partition per node.
    // If we call NUM_NODES, then all replicas across all nodes are covered.
    HashMap<Long, List<String>> replicas = new HashMap<>();
    IntStream.rangeClosed(1,NUM_NODES).forEach(i -> rthread.selectReplicas(nodes).forEach((dnode, rlist) -> rlist.forEach(r -> {
      long pid = r.getReplicaId().getPartitionId().getId();
      List dlist = replicas.getOrDefault(pid, new ArrayList<>());
      dlist.add(dnode.getHostname());
      replicas.putIfAbsent(pid, dlist);
    })));

    // Check that all replicas are covered, the replicas are picked in lexicographical order
    replicas.keySet().forEach(pid -> {
      List<String> dlist = replicas.get(pid);
      List<String> slist = dlist.stream().sorted().collect(Collectors.toList());
      if (dlist.size() != NUM_NODES) {
        logger.error("Insufficient replicas for partition {}, expected {} replicas, but found only {} which are {}",
            pid, NUM_NODES, dlist.size(), String.join(", ", dlist));
        Assert.assertTrue(false);
      }
      if (!slist.equals(dlist)) {
        logger.error("Replicas are not sorted for partition {}, original list = [{}], sorted list = [{}]",
            pid, String.join(", ", dlist), String.join(", ", slist));
        Assert.assertTrue(false);
      }
    });
  }
Tests thread-local metadata cache by fetching blob metadata from azure
@throws CloudStorageException
ambry-cloud/src/test/java/com/github/ambry/cloud/VcrReplicaThreadTest.java