TestLocalExchange.java 文件源码

/** Helper method to verify the number of PartitionSenders in a given fragment endpoint assignments */
private static void verifyAssignment(List<Integer> fragmentList,
    ArrayListMultimap<Integer, DrillbitEndpoint> partitionSenderMap) {

  // We expect at least one entry the list
  assertTrue(fragmentList.size() > 0);

  for(Integer majorFragmentId : fragmentList) {
    // we expect the fragment that has DeMux/HashToRandom as sending exchange to have parallelization with not more
    // than the number of nodes in the cluster and each node in the cluster can have at most one assignment
    List<DrillbitEndpoint> assignments = partitionSenderMap.get(majorFragmentId);
    assertNotNull(assignments);
    assertTrue(assignments.size() > 0);
    assertTrue(String.format("Number of partition senders in major fragment [%d] is more than expected", majorFragmentId), CLUSTER_SIZE >= assignments.size());

    // Make sure there are no duplicates in assigned endpoints (i.e at most one partition sender per endpoint)
    assertTrue("Some endpoints have more than one fragment that has ParitionSender", ImmutableSet.copyOf(assignments).size() == assignments.size());
  }
}