import org.apache.helix.model.ExternalView; //导入依赖的package包/类
private void reconcileViews(List<String> resourceList) throws Exception {
  for (String resource : resourceList) {
    LOG.info("Reconciling resource " + resource);
    ExternalView externalView = helixAdmin.getResourceExternalView(this.clusterName,
        resource);
    if (externalView == null) {
      LOG.warn("External view null for " + resource + " - Skipping");
      continue;
    }
    ViewInfo viewInfo = this.zkManager.getViewInfo(resource);
    if (viewInfo == null) {
      LOG.info("Compressed view does not exist for " + resource);
      this.zkManager.setViewInfo(new ViewInfo(externalView));
    } else {
      ViewInfo currentViewInfo = new ViewInfo(externalView);
      if (!currentViewInfo.equals(viewInfo)) {
        LOG.info("Mismatch b/w external view & compressed view for " + resource);
        this.zkManager.setViewInfo(currentViewInfo);
      } else {
        LOG.info("External view & compressed view match for " + resource);
      }
    }
  }
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
private ViewInfo createViewInfo(Map<String, List<String>> partitionHostMap) {
  ExternalView externalView = new ExternalView(RESOURCE);
  for (Map.Entry<String, List<String>> entry : partitionHostMap.entrySet()) {
    String host = entry.getKey();
    for (String partition : entry.getValue()) {
      String partitionInHelix = RESOURCE + "$" + partition;
      Map<String, String> stateMap = externalView.getStateMap(partitionInHelix);
      if (stateMap == null) {
        stateMap = Maps.newHashMap();
        stateMap.put(host, "ONLINE");
        externalView.setStateMap(partitionInHelix, stateMap);
      } else {
        stateMap.put(host, "ONLINE");
      }
    }
  }
  return new ViewInfo(externalView);
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
private void refresh() {
  LOGGER.info("HelixBrokerRoutingTable.onExternalViewChange");

  PropertyKey externalViews = keyBuilder.externalViews();
  HelixDataAccessor helixDataAccessor = _helixManager.getHelixDataAccessor();
  List<ExternalView> externalViewList = helixDataAccessor.getChildValues(externalViews);

  List<InstanceConfig> instanceConfigList = helixDataAccessor.getChildValues(keyBuilder.instanceConfigs());

  Set<String> servingClusterList = getServingDataResource(externalViewList);
  for (ExternalView externalView : externalViewList) {
    String resourceName = externalView.getResourceName();
    if (servingClusterList.contains(resourceName)) {
      LOGGER.info("Trying to update ExternalView for data resource : " + resourceName + ", ExternalView: "
          + externalView);
      _helixExternalViewBasedRouting.markDataResourceOnline(
          resourceName,
          HelixHelper.getExternalViewForResouce(_helixManager.getClusterManagmentTool(),
              _helixManager.getClusterName(), resourceName), instanceConfigList);
    }
  }
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
private Set<String> getServingDataResource(List<ExternalView> externalViewList) {
  Set<String> servingDataResourceSet = new HashSet<String>();
  for (ExternalView externalView : externalViewList) {
    if (externalView.getResourceName().equals(CommonConstants.Helix.BROKER_RESOURCE_INSTANCE)) {
      Set<String> dataResources = externalView.getPartitionSet();
      for (String dataResource : dataResources) {
        Map<String, String> dataResourceToServingBrokerMap = externalView.getStateMap(dataResource);
        if (dataResourceToServingBrokerMap.containsKey(_instanceId)
            && "ONLINE".equals(dataResourceToServingBrokerMap.get(_instanceId))) {
          servingDataResourceSet.add(dataResource);
        }
      }
    }
  }
  LOGGER.info("Current serving data resource : " + Arrays.toString(servingDataResourceSet.toArray(new String[0])));
  return servingDataResourceSet;
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
private Map<String, Integer> getStateCountMap(ExternalView externalView) {
  Map<String, Integer> stateCountMap = new HashMap<String, Integer>();
  if (externalView == null) {
    return stateCountMap;
  }
  for (String slice : externalView.getPartitionSet()) {
    Map<String, String> stateMap = externalView.getStateMap(slice);
    for (String instance : stateMap.keySet()) {
      String state = stateMap.get(instance);
      if (!stateCountMap.containsKey(state)) {
        stateCountMap.put(state, 0);
      }
      stateCountMap.put(state, stateCountMap.get(state) + 1);
    }
  }
  return stateCountMap;
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
private static void printState(String msg) {
  System.out.println("CLUSTER STATE: " + msg);
  ExternalView resourceExternalView = admin.getResourceExternalView(CLUSTER_NAME, RESOURCE_NAME);
  TreeSet<String> sortedSet = new TreeSet<String>(resourceExternalView.getPartitionSet());
  StringBuilder sb = new StringBuilder("\t\t");
  for (int i = 0; i < NUM_NODES; i++) {
    sb.append(INSTANCE_CONFIG_LIST.get(i).getInstanceName()).append("\t");
  }
  System.out.println(sb);
  for (String partitionName : sortedSet) {
    sb.delete(0, sb.length() - 1);
    sb.append(partitionName).append("\t");
    for (int i = 0; i < NUM_NODES; i++) {
      Map<String, String> stateMap = resourceExternalView.getStateMap(partitionName);
      if (stateMap != null && stateMap.containsKey(INSTANCE_CONFIG_LIST.get(i).getInstanceName())) {
        sb.append(stateMap.get(INSTANCE_CONFIG_LIST.get(i).getInstanceName()).charAt(0)).append(
            "\t\t");
      } else {
        sb.append("-").append("\t\t");
      }
    }
    System.out.println(sb);
  }
  System.out.println("###################################################################");
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
private boolean isLargeCluster(ExternalView externalView) {
  // Check if the number of replicas is sufficient to treat it as a large cluster
  final String helixReplicaCount = externalView.getRecord().getSimpleField("REPLICAS");
  final int replicaCount;

  try {
    replicaCount = Integer.parseInt(helixReplicaCount);
  } catch (Exception e) {
    LOGGER.warn("Failed to parse the replica count ({}) from external view of table {}", helixReplicaCount,
        externalView.getResourceName());
    return false;
  }

  if (replicaCount < _minReplicaCountForLargeCluster) {
    return false;
  }

  // Check if the server count is high enough to count as a large cluster
  final Set<String> instanceSet = new HashSet<>();
  for (String partition : externalView.getPartitionSet()) {
    instanceSet.addAll(externalView.getStateMap(partition).keySet());
  }

  return _minServerCountForLargeCluster <= instanceSet.size();
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
public void markDataResourceOnline(TableConfig tableConfig, ExternalView externalView,
    List<InstanceConfig> instanceConfigList) {
  String tableName = tableConfig.getTableName();

  RoutingTableBuilder routingTableBuilder = _routingTableBuilderFactory.createRoutingTableBuilder(tableConfig);
  routingTableBuilder.init(_configuration, tableConfig, _propertyStore);
  LOGGER.info("Initialized routingTableBuilder: {} for table {}", routingTableBuilder.getClass().getName(), tableName);
  _routingTableBuilderMap.put(tableName, routingTableBuilder);

  // Build the routing table
  if (externalView == null) {
    // It is possible for us to get a request to serve a table for which there is no external view. In this case, just
    // keep a bogus last seen external view version to force a rebuild the next time we see an external view.
    _lastKnownExternalViewVersionMap.put(tableName, INVALID_EXTERNAL_VIEW_VERSION);
    return;
  }
  buildRoutingTable(tableName, externalView, instanceConfigList);
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
@Test (enabled = false)
public void testBatchEnableDisable() throws InterruptedException {
  _gSetupTool.getClusterManagementTool().enableInstance(CLUSTER_NAME,
      Arrays.asList(_participants[0].getInstanceName(), _participants[1].getInstanceName()),
      false);
  Thread.sleep(2000);

  ExternalView externalView = _gSetupTool.getClusterManagementTool()
      .getResourceExternalView(CLUSTER_NAME, WorkflowGenerator.DEFAULT_TGT_DB);
  Assert.assertEquals(externalView.getRecord().getMapFields().size(), _numParitions);
  for (Map<String, String> stateMap : externalView.getRecord().getMapFields().values()) {
    Assert.assertTrue(!stateMap.keySet().contains(_participants[0].getInstanceName()));
    Assert.assertTrue(!stateMap.keySet().contains(_participants[1].getInstanceName()));
  }
  _gSetupTool.getClusterManagementTool().enableInstance(CLUSTER_NAME,
      Arrays.asList(_participants[0].getInstanceName(), _participants[1].getInstanceName()),
      true);
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
@Test (enabled = false)
public void testOldDisableBatchEnable() throws InterruptedException {
  _gSetupTool.getClusterManagementTool()
      .enableInstance(CLUSTER_NAME, _participants[0].getInstanceName(), false);
  _gSetupTool.getClusterManagementTool().enableInstance(CLUSTER_NAME,
      Arrays.asList(_participants[0].getInstanceName(), _participants[1].getInstanceName()),
      true);
  Thread.sleep(2000);

  ExternalView externalView = _gSetupTool.getClusterManagementTool()
      .getResourceExternalView(CLUSTER_NAME, WorkflowGenerator.DEFAULT_TGT_DB);
  Assert.assertEquals(externalView.getRecord().getMapFields().size(), _numParitions);
  int numOfFirstHost = 0;
  for (Map<String, String> stateMap : externalView.getRecord().getMapFields().values()) {
    if (stateMap.keySet().contains(_participants[0].getInstanceName())) {
      numOfFirstHost++;
    }
  }
  Assert.assertTrue(numOfFirstHost > 0);
  _gSetupTool.getClusterManagementTool()
      .enableInstance(CLUSTER_NAME, _participants[0].getInstanceName(), true);
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
@Test (enabled = false)
public void testBatchDisableOldEnable() throws InterruptedException {
  _gSetupTool.getClusterManagementTool().enableInstance(CLUSTER_NAME,
      Arrays.asList(_participants[0].getInstanceName(), _participants[1].getInstanceName()),
      false);
  _gSetupTool.getClusterManagementTool()
      .enableInstance(CLUSTER_NAME, _participants[0].getInstanceName(), true);
  Thread.sleep(2000);

  ExternalView externalView = _gSetupTool.getClusterManagementTool()
      .getResourceExternalView(CLUSTER_NAME, WorkflowGenerator.DEFAULT_TGT_DB);
  Assert.assertEquals(externalView.getRecord().getMapFields().size(), _numParitions);
  int numOfFirstHost = 0;
  for (Map<String, String> stateMap : externalView.getRecord().getMapFields().values()) {
    if (stateMap.keySet().contains(_participants[0].getInstanceName())) {
      numOfFirstHost++;
    }
    Assert.assertTrue(!stateMap.keySet().contains(_participants[1].getInstanceName()));
  }
  Assert.assertTrue(numOfFirstHost > 0);
  _gSetupTool.getClusterManagementTool().enableInstance(CLUSTER_NAME,
      Arrays.asList(_participants[0].getInstanceName(), _participants[1].getInstanceName()),
      true);
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
@Test
public void testCancellationWhenDisableResource() throws InterruptedException {
  // Enable cancellation
  ClusterConfig clusterConfig = _configAccessor.getClusterConfig(CLUSTER_NAME);
  clusterConfig.stateTransitionCancelEnabled(true);
  _configAccessor.setClusterConfig(CLUSTER_NAME, clusterConfig);

  // Wait for assignment done
  Thread.sleep(2000);

  // Disable the resource
  _setupTool.getClusterManagementTool()
      .enableResource(CLUSTER_NAME, WorkflowGenerator.DEFAULT_TGT_DB, false);


  // Wait for pipeline reaching final stage
  Thread.sleep(2000L);
  ExternalView externalView = _setupTool.getClusterManagementTool()
      .getResourceExternalView(CLUSTER_NAME, WorkflowGenerator.DEFAULT_TGT_DB);
  for (String partition : externalView.getPartitionSet()) {
    for (String currentState : externalView.getStateMap(partition).values()) {
      Assert.assertEquals(currentState, "OFFLINE");
    }
  }
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
/**
 * return true if IdealState = ExternalView
 * @return
 */
public int isStable(String tableName) {
  IdealState idealState = helixAdmin.getResourceIdealState(clusterName, tableName);
  ExternalView externalView = helixAdmin.getResourceExternalView(clusterName, tableName);
  Map<String, Map<String, String>> mapFieldsIS = idealState.getRecord().getMapFields();
  Map<String, Map<String, String>> mapFieldsEV = externalView.getRecord().getMapFields();
  int numDiff = 0;
  for (String segment : mapFieldsIS.keySet()) {
    Map<String, String> mapIS = mapFieldsIS.get(segment);
    Map<String, String> mapEV = mapFieldsEV.get(segment);

    for (String server : mapIS.keySet()) {
      String state = mapIS.get(server);
      if (mapEV == null || mapEV.get(server) == null || !mapEV.get(server).equals(state)) {
        LOGGER.info("Mismatch: segment" + segment + " server:" + server + " state:" + state);
        numDiff = numDiff + 1;
      }
    }
  }
  return numDiff;
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
private void startParticipant(
    MockParticipantManager participant, Map<String, MasterSlaveSMD.States> affectedPartitions)
    throws InterruptedException {
  String instance = participant.getInstanceName();
  participant.syncStart();

  Thread.sleep(2000);

  ExternalView externalView = _accessor.getProperty(_keyBuilder.externalView(DB_NAME));
  // Everything back to the initial state
  for (Map.Entry<String, MasterSlaveSMD.States> entry : affectedPartitions.entrySet()) {
    Map<String, String> stateMap = externalView.getStateMap(entry.getKey());
    Assert.assertEquals(stateMap.size(), REPLICA_NUMBER);

    Assert.assertTrue(stateMap.containsKey(instance));
    Assert.assertEquals(stateMap.get(instance), entry.getValue().toString());
  }
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
/**
 * Compute the table of a sorted list of segments grouped by Kafka partition.
 *
 * @param externalView helix external view.
 * @return map of Kafka partition to sorted set of segment names.
 */
public static Map<String, SortedSet<SegmentName>> getSortedSegmentsByKafkaPartition(ExternalView externalView) {
  Map<String, SortedSet<SegmentName>> sortedSegmentsByKafkaPartition = new HashMap<>();
  for (String helixPartitionName : externalView.getPartitionSet()) {
    // Ignore segments that are not low level consumer segments
    if (!SegmentName.isLowLevelConsumerSegmentName(helixPartitionName)) {
      continue;
    }

    final LLCSegmentName segmentName = new LLCSegmentName(helixPartitionName);
    String kafkaPartitionName = segmentName.getPartitionRange();
    SortedSet<SegmentName> segmentsForPartition = sortedSegmentsByKafkaPartition.get(kafkaPartitionName);

    // Create sorted set if necessary
    if (segmentsForPartition == null) {
      segmentsForPartition = new TreeSet<>();

      sortedSegmentsByKafkaPartition.put(kafkaPartitionName, segmentsForPartition);
    }

    segmentsForPartition.add(segmentName);
  }
  return sortedSegmentsByKafkaPartition;
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
/**
 * Validate instances for each partition is on different zone and with necessary tagged instances.
 */
private void validateZoneAndTagIsolation(IdealState is, ExternalView ev, int expectedReplica) {
  String tag = is.getInstanceGroupTag();
  for (String partition : is.getPartitionSet()) {
    Set<String> assignedZones = new HashSet<String>();

    Map<String, String> assignmentMap = ev.getRecord().getMapField(partition);
    Set<String> instancesInEV = assignmentMap.keySet();
    // TODO: preference List is not persisted in IS.
    //Assert.assertEquals(instancesInEV, instancesInIs);
    for (String instance : instancesInEV) {
      assignedZones.add(_nodeToZoneMap.get(instance));
      if (tag != null) {
        InstanceConfig config =
            _setupTool.getClusterManagementTool().getInstanceConfig(CLUSTER_NAME, instance);
        Assert.assertTrue(config.containsTag(tag));
      }
    }
    Assert.assertEquals(assignedZones.size(), expectedReplica);
  }
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
/**
 * The partition movement should be delayed (not happen immediately) after one single node is disabled.
 * Delay is enabled by default, delay time is set in IdealState.
 * @throws Exception
 */
@Test
@Override
public void testDelayedPartitionMovement() throws Exception {
  Map<String, ExternalView> externalViewsBefore = createTestDBs(1000000);

  // Disable one node, no partition should be moved.
  String instance = _participants.get(0).getInstanceName();
  enableInstance(instance, false);

  Thread.sleep(300);
  Assert.assertTrue(_clusterVerifier.verify());

  for (String db : _testDBs) {
    ExternalView ev =
        _setupTool.getClusterManagementTool().getResourceExternalView(CLUSTER_NAME, db);
    IdealState is = _setupTool.getClusterManagementTool().getResourceIdealState(CLUSTER_NAME, db);
    validateMinActiveAndTopStateReplica(is, ev, _minActiveReplica, NUM_NODE);
    validateNoPartitionMove(is, externalViewsBefore.get(db), ev, instance, true);
  }
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
@Test(dependsOnMethods = {"testDelayedPartitionMovement"})
@Override
public void testDelayedPartitionMovementWithClusterConfigedDelay() throws Exception {
  setDelayTimeInCluster(_gZkClient, CLUSTER_NAME, 1000000);

  Map<String, ExternalView> externalViewsBefore = createTestDBs(-1);

  // Disable one node, no partition should be moved.
  String instance = _participants.get(0).getInstanceName();
  enableInstance(instance, false);

  Thread.sleep(100);
  Assert.assertTrue(_clusterVerifier.verify());

  for (String db : _testDBs) {
    ExternalView ev =
        _setupTool.getClusterManagementTool().getResourceExternalView(CLUSTER_NAME, db);
    IdealState is = _setupTool.getClusterManagementTool().getResourceIdealState(CLUSTER_NAME, db);
    validateMinActiveAndTopStateReplica(is, ev, _minActiveReplica, NUM_NODE);
    validateNoPartitionMove(is, externalViewsBefore.get(db), ev,
        _participants.get(0).getInstanceName(), true);
  }

  setDelayTimeInCluster(_gZkClient, CLUSTER_NAME, -1);
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
@Override
public void computeRoutingTableFromExternalView(String tableName, ExternalView externalView,
    List<InstanceConfig> instanceConfigs) {
  Set<String> segmentSet = externalView.getPartitionSet();
  for (String segmentName : segmentSet) {
    if (SegmentName.isHighLevelConsumerSegmentName(segmentName)) {
      _hasHLC = true;
    }
    if (SegmentName.isLowLevelConsumerSegmentName(segmentName)) {
      _hasLLC = true;
    }
  }
  if (_hasHLC) {
    _realtimeHLCRoutingTableBuilder.computeRoutingTableFromExternalView(tableName, externalView, instanceConfigs);
  }
  if (_hasLLC) {
    _realtimeLLCRoutingTableBuilder.computeRoutingTableFromExternalView(tableName, externalView, instanceConfigs);
  }
}
 

import org.apache.helix.model.ExternalView; //导入依赖的package包/类
/**
 * The partititon should be moved to other nodes after the delay time
 */
@Test (dependsOnMethods = {"testMinimalActiveReplicaMaintain"})
public void testPartitionMovementAfterDelayTime() throws Exception {
  enablePersistBestPossibleAssignment(_gZkClient, CLUSTER_NAME, true);

  long delay = 4000;
  Map<String, ExternalView> externalViewsBefore = createTestDBs(delay);
  validateDelayedMovements(externalViewsBefore);

  Thread.sleep(delay + 200);
  Assert.assertTrue(_clusterVerifier.verify());
  // after delay time, it should maintain required number of replicas.
  for (String db : _testDBs) {
    ExternalView ev =
        _setupTool.getClusterManagementTool().getResourceExternalView(CLUSTER_NAME, db);
    IdealState is = _setupTool.getClusterManagementTool().getResourceIdealState(CLUSTER_NAME, db);
    validateMinActiveAndTopStateReplica(is, ev, _replica, NUM_NODE);
  }
}