import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
@Override
public List<String> getAssignedInstances(HelixAdmin helixAdmin, ZkHelixPropertyStore<ZNRecord> propertyStore,
    String helixClusterName, SegmentMetadata segmentMetadata, int numReplicas, String tenantName) {
  String serverTenantName = ControllerTenantNameBuilder.getOfflineTenantNameForTenant(tenantName);
  final Random random = new Random(System.currentTimeMillis());

  List<String> allInstanceList =
      HelixHelper.getEnabledInstancesWithTag(helixAdmin, helixClusterName, serverTenantName);
  List<String> selectedInstanceList = new ArrayList<>();
  for (int i = 0; i < numReplicas; ++i) {
    final int idx = random.nextInt(allInstanceList.size());
    selectedInstanceList.add(allInstanceList.get(idx));
    allInstanceList.remove(idx);
  }
  LOGGER.info("Segment assignment result for : " + segmentMetadata.getName() + ", in resource : "
      + segmentMetadata.getTableName() + ", selected instances: " + Arrays.toString(selectedInstanceList.toArray()));

  return selectedInstanceList;
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
public static List<RealtimeSegmentZKMetadata> getRealtimeSegmentZKMetadataListForTable(
    ZkHelixPropertyStore<ZNRecord> propertyStore, String resourceName) {
  List<RealtimeSegmentZKMetadata> resultList = new ArrayList<>();
  if (propertyStore == null) {
    return resultList;
  }
  String realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(resourceName);
  if (propertyStore.exists(constructPropertyStorePathForResource(realtimeTableName), AccessOption.PERSISTENT)) {
    List<ZNRecord> znRecordList =
        propertyStore.getChildren(constructPropertyStorePathForResource(realtimeTableName), null,
            AccessOption.PERSISTENT);
    if (znRecordList != null) {
      for (ZNRecord record : znRecordList) {
        resultList.add(new RealtimeSegmentZKMetadata(record));
      }
    }
  }
  return resultList;
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
public static IdealState buildInitialRealtimeIdealStateFor(String realtimeTableName,
    AbstractTableConfig realtimeTableConfig, HelixAdmin helixAdmin, String helixClusterName,
    ZkHelixPropertyStore<ZNRecord> zkHelixPropertyStore) {
  KafkaStreamMetadata kafkaStreamMetadata =
      new KafkaStreamMetadata(realtimeTableConfig.getIndexingConfig().getStreamConfigs());
  String realtimeServerTenant =
      ControllerTenantNameBuilder.getRealtimeTenantNameForTenant(realtimeTableConfig.getTenantConfig().getServer());
  switch (kafkaStreamMetadata.getConsumerType()) {
    case highLevel:
      IdealState idealState =
          buildInitialKafkaHighLevelConsumerRealtimeIdealStateFor(realtimeTableName, helixAdmin, helixClusterName,
              zkHelixPropertyStore);
      List<String> realtimeInstances = helixAdmin.getInstancesInClusterWithTag(helixClusterName, realtimeServerTenant);
      if (realtimeInstances.size() % Integer.parseInt(realtimeTableConfig.getValidationConfig().getReplication()) != 0) {
        throw new RuntimeException("Number of instance in current tenant should be an integer multiples of the number of replications");
      }
      setupInstanceConfigForKafkaHighLevelConsumer(realtimeTableName, realtimeInstances.size(),
          Integer.parseInt(realtimeTableConfig.getValidationConfig().getReplication()), realtimeTableConfig
              .getIndexingConfig().getStreamConfigs(), zkHelixPropertyStore, realtimeInstances);
      return idealState;
    case simple:
    default:
      throw new UnsupportedOperationException("Not support kafka consumer type: "
          + kafkaStreamMetadata.getConsumerType());
  }
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
public HelixExternalViewBasedRouting(RoutingTableBuilder defaultOfflineRoutingTableBuilder,
    RoutingTableBuilder defaultRealtimeRoutingTableBuilder, Map<String, RoutingTableBuilder> routingTableBuilderMap,
    ZkHelixPropertyStore<ZNRecord> propertyStore) {
  _timeBoundaryService = new HelixExternalViewBasedTimeBoundaryService(propertyStore);
  if (defaultOfflineRoutingTableBuilder != null) {
    _defaultOfflineRoutingTableBuilder = defaultOfflineRoutingTableBuilder;
  } else {
    _defaultOfflineRoutingTableBuilder = new BalancedRandomRoutingTableBuilder();
  }
  if (defaultRealtimeRoutingTableBuilder != null) {
    _defaultRealtimeRoutingTableBuilder = defaultRealtimeRoutingTableBuilder;
  } else {
    _defaultRealtimeRoutingTableBuilder = new KafkaHighLevelConsumerBasedRoutingTableBuilder();
  }
  if (routingTableBuilderMap != null) {
    _routingTableBuilderMap = routingTableBuilderMap;
  } else {
    _routingTableBuilderMap = new HashMap<String, RoutingTableBuilder>();
  }
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
private void addingSegmentsToPropertyStore(int numSegments, ZkHelixPropertyStore<ZNRecord> propertyStore,
    String tableName) {
  for (int i = 0; i < numSegments; ++i) {
    OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
    offlineSegmentZKMetadata.setSegmentName(tableName + "_" + System.currentTimeMillis() + "_" + i);
    offlineSegmentZKMetadata.setTimeUnit(TimeUnit.DAYS);
    offlineSegmentZKMetadata.setEndTime(i);
    offlineSegmentZKMetadata.setCrc(-1);
    offlineSegmentZKMetadata.setCreationTime(-1);
    offlineSegmentZKMetadata.setStartTime(i - 1);
    offlineSegmentZKMetadata.setIndexVersion("0");
    offlineSegmentZKMetadata.setPushTime(i + 5);
    offlineSegmentZKMetadata.setTableName(tableName);
    offlineSegmentZKMetadata.setSegmentType(SegmentType.OFFLINE);
    ZKMetadataProvider.setOfflineSegmentZKMetadata(_propertyStore, offlineSegmentZKMetadata);
  }
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
public static List<OfflineSegmentZKMetadata> getOfflineSegmentZKMetadataListForTable(ZkHelixPropertyStore<ZNRecord> propertyStore, String tableName) {
  List<OfflineSegmentZKMetadata> resultList = new ArrayList<OfflineSegmentZKMetadata>();
  if (propertyStore == null) {
    return resultList;
  }
  String offlineTableName = TableNameBuilder.OFFLINE_TABLE_NAME_BUILDER.forTable(tableName);
  if (propertyStore.exists(constructPropertyStorePathForResource(offlineTableName), AccessOption.PERSISTENT)) {
    List<ZNRecord> znRecordList = propertyStore.getChildren(constructPropertyStorePathForResource(offlineTableName), null, AccessOption.PERSISTENT);
    if (znRecordList != null) {
      for (ZNRecord record : znRecordList) {
        resultList.add(new OfflineSegmentZKMetadata(record));
      }
    }
  }
  return resultList;
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
public static List<RealtimeSegmentZKMetadata> getRealtimeSegmentZKMetadataListForTable(ZkHelixPropertyStore<ZNRecord> propertyStore, String resourceName) {
  List<RealtimeSegmentZKMetadata> resultList = new ArrayList<RealtimeSegmentZKMetadata>();
  if (propertyStore == null) {
    return resultList;
  }
  String realtimeTableName = TableNameBuilder.REALTIME_TABLE_NAME_BUILDER.forTable(resourceName);
  if (propertyStore.exists(constructPropertyStorePathForResource(realtimeTableName), AccessOption.PERSISTENT)) {
    List<ZNRecord> znRecordList = propertyStore.getChildren(constructPropertyStorePathForResource(realtimeTableName), null, AccessOption.PERSISTENT);
    if (znRecordList != null) {
      for (ZNRecord record : znRecordList) {
        resultList.add(new RealtimeSegmentZKMetadata(record));
      }
    }
  }
  return resultList;
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
@Override
public boolean execute()
    throws Exception {
  if (!_validateTableConfig && !_validateSchema) {
    throw new RuntimeException("Need to specify at least one of -schema and -tableConfig");
  }

  LOGGER.info("Connecting to Zookeeper: {}, cluster: ", _zkAddress, _clusterName);
  ZNRecordSerializer serializer = new ZNRecordSerializer();
  String path = PropertyPathConfig.getPath(PropertyType.PROPERTYSTORE, _clusterName);
  _helixPropertyStore = new ZkHelixPropertyStore<>(_zkAddress, serializer, path);

  LOGGER.info("\n\n-------------------- Starting Validation --------------------");
  if (_validateTableConfig) {
    validateTableConfig();
  }
  if (_validateSchema) {
    validateSchema();
  }

  return true;
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
public PinotZKChanger(String zkAddress, String clusterName) {
  this.clusterName = clusterName;
  helixAdmin = new ZKHelixAdmin(zkAddress);
  helixManager =
      HelixManagerFactory
          .getZKHelixManager(clusterName, "PinotNumReplicaChanger", InstanceType.ADMINISTRATOR, zkAddress);
  try {
    helixManager.connect();
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
  ZNRecordSerializer serializer = new ZNRecordSerializer();
  String path = PropertyPathConfig.getPath(PropertyType.PROPERTYSTORE, clusterName);
  propertyStore = new ZkHelixPropertyStore<>(zkAddress, serializer, path);
  objectMapper = new ObjectMapper();
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
@Override
public List<String> getAssignedInstances(HelixAdmin helixAdmin, ZkHelixPropertyStore<ZNRecord> propertyStore,
    String helixClusterName, SegmentMetadata segmentMetadata, int numReplicas, String tenantName) {
  String serverTenantName = ControllerTenantNameBuilder.getOfflineTenantNameForTenant(tenantName);

  List<String> allInstances = HelixHelper.getEnabledInstancesWithTag(helixAdmin, helixClusterName, serverTenantName);
  List<String> selectedInstanceList = new ArrayList<>();
  if (segmentMetadata.getShardingKey() != null) {
    for (String instance : allInstances) {
      if (HelixHelper.getInstanceConfigsMapFor(instance, helixClusterName, helixAdmin)
          .get("shardingKey")
          .equalsIgnoreCase(segmentMetadata.getShardingKey())) {
        selectedInstanceList.add(instance);
      }
    }
    LOGGER.info("Segment assignment result for : " + segmentMetadata.getName() + ", in resource : "
        + segmentMetadata.getTableName() + ", selected instances: " + Arrays.toString(
        selectedInstanceList.toArray()));
    return selectedInstanceList;
  } else {
    throw new RuntimeException("Segment missing sharding key!");
  }
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
protected PinotLLCRealtimeSegmentManager(HelixAdmin helixAdmin, String clusterName, HelixManager helixManager,
    ZkHelixPropertyStore propertyStore, PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
    ControllerMetrics controllerMetrics) {
  _helixAdmin = helixAdmin;
  _helixManager = helixManager;
  _propertyStore = propertyStore;
  _helixResourceManager = helixResourceManager;
  _clusterName = clusterName;
  _controllerConf = controllerConf;
  _controllerMetrics = controllerMetrics;
  _idealstateUpdateLocks = new Lock[NUM_LOCKS];
  for (int i = 0; i < NUM_LOCKS; i++) {
    _idealstateUpdateLocks[i] = new ReentrantLock();
  }
  _tableConfigCache = new TableConfigCache(_propertyStore);
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
public static IdealState buildInitialHighLevelRealtimeIdealStateFor(String realtimeTableName,
    TableConfig realtimeTableConfig, HelixAdmin helixAdmin, String helixClusterName,
    ZkHelixPropertyStore<ZNRecord> zkHelixPropertyStore) {
  String realtimeServerTenant =
      ControllerTenantNameBuilder.getRealtimeTenantNameForTenant(realtimeTableConfig.getTenantConfig().getServer());
  final List<String> realtimeInstances = helixAdmin.getInstancesInClusterWithTag(helixClusterName,
      realtimeServerTenant);
  IdealState idealState = buildEmptyKafkaConsumerRealtimeIdealStateFor(realtimeTableName, 1);
  if (realtimeInstances.size() % Integer.parseInt(realtimeTableConfig.getValidationConfig().getReplication()) != 0) {
    throw new RuntimeException(
        "Number of instance in current tenant should be an integer multiples of the number of replications");
  }
  setupInstanceConfigForKafkaHighLevelConsumer(realtimeTableName, realtimeInstances.size(),
      Integer.parseInt(realtimeTableConfig.getValidationConfig().getReplication()),
      realtimeTableConfig.getIndexingConfig().getStreamConfigs(), zkHelixPropertyStore, realtimeInstances);
  return idealState;
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
public void testBulkDeleteWithFailures(boolean useSet) throws Exception {
  HelixAdmin helixAdmin =  makeHelixAdmin();
  ZkHelixPropertyStore<ZNRecord> propertyStore = makePropertyStore();
  FakeDeletionManager deletionManager = new FakeDeletionManager(helixAdmin, propertyStore);
  Collection<String> segments;
  if (useSet) {
    segments = new HashSet<String>();
  } else {
    segments = new ArrayList<String>();
  }

  segments.addAll(segmentsThatShouldBeDeleted());
  segments.addAll(segmentsInIdealStateOrExtView());
  segments.addAll(segmentsFailingPropStore());
  deletionManager.deleteSegmentsFromPropertyStoreAndLocal(tableName, segments);

  Assert.assertTrue(deletionManager.segmentsToRetry.containsAll(segmentsFailingPropStore()));
  Assert.assertTrue(deletionManager.segmentsToRetry.containsAll(segmentsInIdealStateOrExtView()));
  Assert.assertTrue(deletionManager.segmentsRemovedFromStore.containsAll(segmentsThatShouldBeDeleted()));
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
@Override
public void init(Configuration configuration, TableConfig tableConfig, ZkHelixPropertyStore<ZNRecord> propertyStore) {
  // TODO jfim This is a broker-level configuration for now, until we refactor the configuration of the routing table to allow per-table routing settings
  if (configuration.containsKey("offlineTargetServerCountPerQuery")) {
    final String targetServerCountPerQuery = configuration.getString("offlineTargetServerCountPerQuery");
    try {
      _targetNumServersPerQuery = Integer.parseInt(targetServerCountPerQuery);
      LOGGER.info("Using offline target server count of {}", _targetNumServersPerQuery);
    } catch (Exception e) {
      LOGGER.warn(
          "Could not get the offline target server count per query from configuration value {}, keeping default value {}",
          targetServerCountPerQuery, _targetNumServersPerQuery, e);
    }
  } else {
    LOGGER.info("Using default value for offline target server count of {}", _targetNumServersPerQuery);
  }
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
@Override
public void init(Configuration configuration, TableConfig tableConfig, ZkHelixPropertyStore<ZNRecord> propertyStore) {
  // TODO jfim This is a broker-level configuration for now, until we refactor the configuration of the routing table to allow per-table routing settings
  if (configuration.containsKey("realtimeTargetServerCountPerQuery")) {
    final String targetServerCountPerQuery = configuration.getString("realtimeTargetServerCountPerQuery");
    try {
      _targetNumServersPerQuery = Integer.parseInt(targetServerCountPerQuery);
      LOGGER.info("Using realtime target server count of {}", _targetNumServersPerQuery);
    } catch (Exception e) {
      LOGGER.warn(
          "Could not get the realtime target server count per query from configuration value {}, keeping default value {}",
          targetServerCountPerQuery, _targetNumServersPerQuery, e);
    }
  } else {
    LOGGER.info("Using default value for realtime target server count of {}", _targetNumServersPerQuery);
  }
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
@Nullable
public static RealtimeSegmentZKMetadata getRealtimeSegmentZKMetadata(
    @Nonnull ZkHelixPropertyStore<ZNRecord> propertyStore, @Nonnull String tableName, @Nonnull String segmentName) {
  String realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(tableName);
  ZNRecord znRecord = propertyStore.get(constructPropertyStorePathForSegment(realtimeTableName, segmentName), null,
      AccessOption.PERSISTENT);
  // It is possible that the segment metadata has just been deleted due to retention.
  if (znRecord == null) {
    return null;
  }
  if (SegmentName.isHighLevelConsumerSegmentName(segmentName)) {
    return new RealtimeSegmentZKMetadata(znRecord);
  } else {
    return new LLCRealtimeSegmentZKMetadata(znRecord);
  }
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
public static List<OfflineSegmentZKMetadata> getOfflineSegmentZKMetadataListForTable(
    ZkHelixPropertyStore<ZNRecord> propertyStore, String tableName) {
  List<OfflineSegmentZKMetadata> resultList = new ArrayList<>();
  if (propertyStore == null) {
    return resultList;
  }
  String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
  if (propertyStore.exists(constructPropertyStorePathForResource(offlineTableName), AccessOption.PERSISTENT)) {
    List<ZNRecord> znRecordList =
        propertyStore.getChildren(constructPropertyStorePathForResource(offlineTableName), null,
            AccessOption.PERSISTENT);
    if (znRecordList != null) {
      for (ZNRecord record : znRecordList) {
        resultList.add(new OfflineSegmentZKMetadata(record));
      }
    }
  }
  return resultList;
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
public static TableDataManager getTableDataManager(@Nonnull TableDataManagerConfig tableDataManagerConfig,
    @Nonnull String instanceId, @Nonnull ZkHelixPropertyStore<ZNRecord> propertyStore,
    @Nonnull ServerMetrics serverMetrics) {
  String tableDataManagerType = tableDataManagerConfig.getTableDataManagerType().toUpperCase();
  TableDataManager tableDataManager;
  switch (tableDataManagerType) {
    case OFFLINE_TABLE_DATA_MANAGER_TYPE:
      tableDataManager = new OfflineTableDataManager();
      break;
    case REALTIME_TABLE_DATA_MANAGER_TYPE:
      tableDataManager = new RealtimeTableDataManager();
      break;
    default:
      throw new UnsupportedOperationException(
          "Unsupported table data manager type: " + tableDataManagerType + " for table: "
              + tableDataManagerConfig.getTableName());
  }
  tableDataManager.init(tableDataManagerConfig, instanceId, propertyStore, serverMetrics);
  return tableDataManager;
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
@Override
public void init(@Nonnull TableDataManagerConfig tableDataManagerConfig, @Nonnull String instanceId,
    @Nonnull ZkHelixPropertyStore<ZNRecord> propertyStore, @Nonnull ServerMetrics serverMetrics) {
  _tableDataManagerConfig = tableDataManagerConfig;
  _instanceId = instanceId;
  _propertyStore = propertyStore;
  _serverMetrics = serverMetrics;

  _tableName = tableDataManagerConfig.getTableName();
  _tableDataDir = tableDataManagerConfig.getDataDir();
  _indexDir = new File(_tableDataDir);
  if (!_indexDir.exists()) {
    Preconditions.checkState(_indexDir.mkdirs());
  }
  _logger = LoggerFactory.getLogger(_tableName + "-" + getClass().getSimpleName());

  doInit();

  _logger.info("Initialized table: {} with data directory: {}", _tableName, _tableDataDir);
}
 

import org.apache.helix.store.zk.ZkHelixPropertyStore; //导入依赖的package包/类
private OfflineTableDataManager makeTestableManager() throws Exception {
  OfflineTableDataManager tableDataManager = new OfflineTableDataManager();
  TableDataManagerConfig config;
  {
    config = mock(TableDataManagerConfig.class);
    when(config.getTableName()).thenReturn(tableName);
    when(config.getDataDir()).thenReturn(_tmpDir.getAbsolutePath());
  }
  tableDataManager.init(config, "dummyInstance", mock(ZkHelixPropertyStore.class),
      new ServerMetrics(new MetricsRegistry()));
  tableDataManager.start();
  Field segsMapField = AbstractTableDataManager.class.getDeclaredField("_segmentsMap");
  segsMapField.setAccessible(true);
  _internalSegMap = (Map<String, OfflineSegmentDataManager>)segsMapField.get(tableDataManager);
  return tableDataManager;
}