@Test public void testLeadControllerAssignment() { for (int nInstances = 1; nInstances <= MAXIMUM_NUMBER_OF_CONTROLLER_INSTANCES; nInstances++) { List<String> instanceNames = new ArrayList<>(nInstances); List<Integer> ports = new ArrayList<>(nInstances); for (int i = 0; i < nInstances; i++) { instanceNames.add(LeadControllerUtils.generateParticipantInstanceId(LOCAL_HOST, i)); ports.add(i); } List<String> partitions = new ArrayList<>(NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE); for (int i = 0; i < NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE; i++) { partitions.add(LeadControllerUtils.generatePartitionName(i)); } LinkedHashMap<String, Integer> states = new LinkedHashMap<>(2); states.put(MasterSlaveSMD.States.OFFLINE.name(), 0); states.put(MasterSlaveSMD.States.SLAVE.name(), LEAD_CONTROLLER_RESOURCE_REPLICA_COUNT - 1); states.put(MasterSlaveSMD.States.MASTER.name(), 1); CrushEdRebalanceStrategy crushEdRebalanceStrategy = new CrushEdRebalanceStrategy(); crushEdRebalanceStrategy.init(LEAD_CONTROLLER_RESOURCE_NAME, partitions, states, Integer.MAX_VALUE); ClusterDataCache clusterDataCache = new ClusterDataCache(); PropertyKey.Builder keyBuilder = new PropertyKey.Builder(getHelixClusterName()); HelixDataAccessor accessor = _helixManager.getHelixDataAccessor(); ClusterConfig clusterConfig = accessor.getProperty(keyBuilder.clusterConfig()); clusterDataCache.setClusterConfig(clusterConfig); Map<String, InstanceConfig> instanceConfigMap = new HashMap<>(nInstances); for (int i = 0; i < nInstances; i++) { String instanceName = instanceNames.get(i); int port = ports.get(i); instanceConfigMap.put(instanceName, new InstanceConfig(instanceName + ", {HELIX_ENABLED=true, HELIX_ENABLED_TIMESTAMP=1559546216610, HELIX_HOST=Controller_localhost, HELIX_PORT=" + port + "}{}{TAG_LIST=[controller]}")); } clusterDataCache.setInstanceConfigMap(instanceConfigMap); ZNRecord znRecord = crushEdRebalanceStrategy .computePartitionAssignment(instanceNames, instanceNames, new HashMap<>(0), clusterDataCache); Assert.assertNotNull(znRecord); Map<String, List<String>> listFields = znRecord.getListFields(); Assert.assertEquals(listFields.size(), NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE); Map<String, Integer> instanceToMasterAssignmentCountMap = new HashMap<>(); int maxCount = 0; for (List<String> assignments : listFields.values()) { Assert.assertEquals(assignments.size(), LEAD_CONTROLLER_RESOURCE_REPLICA_COUNT); if (!instanceToMasterAssignmentCountMap.containsKey(assignments.get(0))) { instanceToMasterAssignmentCountMap.put(assignments.get(0), 1); } else { instanceToMasterAssignmentCountMap .put(assignments.get(0), instanceToMasterAssignmentCountMap.get(assignments.get(0)) + 1); } maxCount = Math.max(instanceToMasterAssignmentCountMap.get(assignments.get(0)), maxCount); } Assert.assertEquals(instanceToMasterAssignmentCountMap.size(), nInstances, "Not all the instances got assigned to the resource!"); for (Integer count : instanceToMasterAssignmentCountMap.values()) { Assert.assertTrue((maxCount - count == 0 || maxCount - count == 1), "Instance assignment isn't distributed"); } } }
public String getHelixClusterName() { return _helixClusterName; }
PinotHelixResourceManager { public String getHelixClusterName() { return _helixClusterName; } }
PinotHelixResourceManager { public String getHelixClusterName() { return _helixClusterName; } PinotHelixResourceManager(String zkURL, String helixClusterName, @Nullable String dataDir,
      long externalViewOnlineToOfflineTimeoutMillis, boolean isSingleTenantCluster, boolean enableBatchMessageMode,
      boolean allowHLCTables);  PinotHelixResourceManager(ControllerConf controllerConf); }
PinotHelixResourceManager { public String getHelixClusterName() { return _helixClusterName; } PinotHelixResourceManager(String zkURL, String helixClusterName, @Nullable String dataDir,
      long externalViewOnlineToOfflineTimeoutMillis, boolean isSingleTenantCluster, boolean enableBatchMessageMode,
      boolean allowHLCTables);  PinotHelixResourceManager(ControllerConf controllerConf); synchronized void start(HelixManager helixZkManager); synchronized void stop(); String getHelixZkURL(); String getHelixClusterName(); SegmentDeletionManager getSegmentDeletionManager(); HelixManager getHelixZkManager(); HelixAdmin getHelixAdmin(); ZkHelixPropertyStore<ZNRecord> getPropertyStore(); List<String> getAllInstances(); List<InstanceConfig> getAllHelixInstanceConfigs(); @Nullable InstanceConfig getHelixInstanceConfig(String instanceId); @Nullable InstanceZKMetadata getInstanceZKMetadata(String instanceId); List<String> getBrokerInstancesFor(String tableName); List<String> getInstancesWithTag(String tag); synchronized PinotResourceManagerResponse addInstance(Instance instance); synchronized PinotResourceManagerResponse updateInstance(String instanceIdToUpdate, Instance newInstance); synchronized PinotResourceManagerResponse updateInstanceTags(String instanceIdToUpdate, String tags); List<String> getAllResources(); List<String> getAllTables(); List<String> getAllOfflineTables(); List<String> getAllRealtimeTables(); List<String> getAllRawTables(); String getActualTableName(String tableName); String getCrypterClassNameFromTableConfig(String tableNameWithType); List<String> getSegmentsFor(String tableNameWithType); @Nullable OfflineSegmentZKMetadata getOfflineSegmentZKMetadata(String tableName, String segmentName); List<OfflineSegmentZKMetadata> getOfflineSegmentMetadata(String tableName); List<RealtimeSegmentZKMetadata> getRealtimeSegmentMetadata(String tableName); synchronized PinotResourceManagerResponse deleteSegments(String tableNameWithType, List<String> segmentNames); synchronized PinotResourceManagerResponse deleteSegment(String tableNameWithType, String segmentName); PinotResourceManagerResponse updateBrokerTenant(Tenant tenant); PinotResourceManagerResponse rebuildBrokerResourceFromHelixTags(String tableNameWithType); PinotResourceManagerResponse rebuildBrokerResource(String tableNameWithType, Set<String> brokerInstances); PinotResourceManagerResponse updateServerTenant(Tenant serverTenant); boolean isBrokerTenantDeletable(String tenantName); boolean isServerTenantDeletable(String tenantName); Set<String> getAllBrokerTenantNames(); Set<String> getAllServerTenantNames(); PinotResourceManagerResponse createServerTenant(Tenant serverTenant); PinotResourceManagerResponse createBrokerTenant(Tenant brokerTenant); PinotResourceManagerResponse deleteOfflineServerTenantFor(String tenantName); PinotResourceManagerResponse deleteRealtimeServerTenantFor(String tenantName); PinotResourceManagerResponse deleteBrokerTenantFor(String tenantName); Set<String> getAllInstancesForServerTenant(List<InstanceConfig> instanceConfigs, String tenantName); Set<String> getAllInstancesForServerTenant(String tenantName); Set<String> getAllInstancesForBrokerTenant(List<InstanceConfig> instanceConfigs, String tenantName); Set<String> getAllInstancesForBrokerTenant(String tenantName); void addSchema(Schema schema, boolean override); void updateSchema(Schema schema, boolean reload); boolean deleteSchema(Schema schema); @Nullable Schema getSchema(String schemaName); @Nullable Schema getTableSchema(String tableName); @Nullable Schema getSchemaForTableConfig(TableConfig tableConfig); List<String> getSchemaNames(); void addTable(TableConfig tableConfig); boolean setZKData(String path, ZNRecord record, int expectedVersion, int accessOption); boolean deleteZKPath(String path); ZNRecord readZKData(String path); List<String> getZKChildren(String path); Map<String, Stat> getZKChildrenStats(String path); Stat getZKStat(String path); void registerPinotLLCRealtimeSegmentManager(PinotLLCRealtimeSegmentManager pinotLLCRealtimeSegmentManager); void updateTableConfig(TableConfig tableConfig); void setExistingTableConfig(TableConfig tableConfig); void updateMetadataConfigFor(String tableName, TableType type, TableCustomConfig newConfigs); void updateSegmentsValidationAndRetentionConfigFor(String tableName, TableType type,
      SegmentsValidationAndRetentionConfig newConfigs); void updateIndexingConfigFor(String tableName, TableType type, IndexingConfig newConfigs); void deleteOfflineTable(String tableName); void deleteRealtimeTable(String tableName); PinotResourceManagerResponse toggleTableState(String tableNameWithType, StateType stateType); void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl); void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
      @Nullable String crypter); @Nullable ZNRecord getSegmentMetadataZnRecord(String tableNameWithType, String segmentName); boolean updateZkMetadata(String offlineTableName, OfflineSegmentZKMetadata segmentMetadata,
      int expectedVersion); boolean updateZkMetadata(String offlineTableName, OfflineSegmentZKMetadata segmentMetadata); void refreshSegment(String offlineTableName, SegmentMetadata segmentMetadata,
      OfflineSegmentZKMetadata offlineSegmentZKMetadata, String downloadUrl, @Nullable String crypter); int reloadAllSegments(String tableNameWithType); int reloadSegment(String tableNameWithType, String segmentName); void sendSegmentRefreshMessage(String tableNameWithType, String segmentName, boolean refreshServerSegment,
      boolean refreshBrokerRouting); void toggleQueryQuotaStateForBroker(String brokerInstanceName, String state); Map<String, List<String>> getServerToSegmentsMap(String tableNameWithType); synchronized Map<String, String> getSegmentsCrcForTable(String tableNameWithType); String buildPathForSegmentMetadata(String tableNameWithType, String segmentName); boolean hasTable(String tableNameWithType); boolean hasOfflineTable(String tableName); boolean hasRealtimeTable(String tableName); @Nullable IdealState getTableIdealState(String tableNameWithType); @Nullable ExternalView getTableExternalView(String tableNameWithType); @Nullable TableConfig getTableConfig(String tableNameWithType); @Nullable TableConfig getOfflineTableConfig(String tableName); @Nullable TableConfig getRealtimeTableConfig(String tableName); @Nullable TableConfig getTableConfig(String tableName, TableType tableType); List<TableConfig> getTableConfigsForSchema(String schemaName); List<String> getServerInstancesForTable(String tableName, TableType tableType); List<String> getBrokerInstancesForTable(String tableName, TableType tableType); PinotResourceManagerResponse enableInstance(String instanceName); PinotResourceManagerResponse disableInstance(String instanceName); PinotResourceManagerResponse dropInstance(String instanceName); RebalanceResult rebalanceTable(String tableNameWithType, Configuration rebalanceConfig); boolean instanceExists(String instanceName); List<String> getOnlineUnTaggedBrokerInstanceList(); List<String> getOnlineUnTaggedServerInstanceList(); List<String> getOnlineInstanceList(); BiMap<String, String> getDataInstanceAdminEndpoints(Set<String> instances); List<String> getExistingTableNamesWithType(String tableName, @Nullable TableType tableType); String startReplaceSegments(String tableNameWithType, List<String> segmentsFrom, List<String> segmentsTo); void endReplaceSegments(String tableNameWithType, String segmentLineageEntryId); }
PinotHelixResourceManager { public String getHelixClusterName() { return _helixClusterName; } PinotHelixResourceManager(String zkURL, String helixClusterName, @Nullable String dataDir,
      long externalViewOnlineToOfflineTimeoutMillis, boolean isSingleTenantCluster, boolean enableBatchMessageMode,
      boolean allowHLCTables);  PinotHelixResourceManager(ControllerConf controllerConf); synchronized void start(HelixManager helixZkManager); synchronized void stop(); String getHelixZkURL(); String getHelixClusterName(); SegmentDeletionManager getSegmentDeletionManager(); HelixManager getHelixZkManager(); HelixAdmin getHelixAdmin(); ZkHelixPropertyStore<ZNRecord> getPropertyStore(); List<String> getAllInstances(); List<InstanceConfig> getAllHelixInstanceConfigs(); @Nullable InstanceConfig getHelixInstanceConfig(String instanceId); @Nullable InstanceZKMetadata getInstanceZKMetadata(String instanceId); List<String> getBrokerInstancesFor(String tableName); List<String> getInstancesWithTag(String tag); synchronized PinotResourceManagerResponse addInstance(Instance instance); synchronized PinotResourceManagerResponse updateInstance(String instanceIdToUpdate, Instance newInstance); synchronized PinotResourceManagerResponse updateInstanceTags(String instanceIdToUpdate, String tags); List<String> getAllResources(); List<String> getAllTables(); List<String> getAllOfflineTables(); List<String> getAllRealtimeTables(); List<String> getAllRawTables(); String getActualTableName(String tableName); String getCrypterClassNameFromTableConfig(String tableNameWithType); List<String> getSegmentsFor(String tableNameWithType); @Nullable OfflineSegmentZKMetadata getOfflineSegmentZKMetadata(String tableName, String segmentName); List<OfflineSegmentZKMetadata> getOfflineSegmentMetadata(String tableName); List<RealtimeSegmentZKMetadata> getRealtimeSegmentMetadata(String tableName); synchronized PinotResourceManagerResponse deleteSegments(String tableNameWithType, List<String> segmentNames); synchronized PinotResourceManagerResponse deleteSegment(String tableNameWithType, String segmentName); PinotResourceManagerResponse updateBrokerTenant(Tenant tenant); PinotResourceManagerResponse rebuildBrokerResourceFromHelixTags(String tableNameWithType); PinotResourceManagerResponse rebuildBrokerResource(String tableNameWithType, Set<String> brokerInstances); PinotResourceManagerResponse updateServerTenant(Tenant serverTenant); boolean isBrokerTenantDeletable(String tenantName); boolean isServerTenantDeletable(String tenantName); Set<String> getAllBrokerTenantNames(); Set<String> getAllServerTenantNames(); PinotResourceManagerResponse createServerTenant(Tenant serverTenant); PinotResourceManagerResponse createBrokerTenant(Tenant brokerTenant); PinotResourceManagerResponse deleteOfflineServerTenantFor(String tenantName); PinotResourceManagerResponse deleteRealtimeServerTenantFor(String tenantName); PinotResourceManagerResponse deleteBrokerTenantFor(String tenantName); Set<String> getAllInstancesForServerTenant(List<InstanceConfig> instanceConfigs, String tenantName); Set<String> getAllInstancesForServerTenant(String tenantName); Set<String> getAllInstancesForBrokerTenant(List<InstanceConfig> instanceConfigs, String tenantName); Set<String> getAllInstancesForBrokerTenant(String tenantName); void addSchema(Schema schema, boolean override); void updateSchema(Schema schema, boolean reload); boolean deleteSchema(Schema schema); @Nullable Schema getSchema(String schemaName); @Nullable Schema getTableSchema(String tableName); @Nullable Schema getSchemaForTableConfig(TableConfig tableConfig); List<String> getSchemaNames(); void addTable(TableConfig tableConfig); boolean setZKData(String path, ZNRecord record, int expectedVersion, int accessOption); boolean deleteZKPath(String path); ZNRecord readZKData(String path); List<String> getZKChildren(String path); Map<String, Stat> getZKChildrenStats(String path); Stat getZKStat(String path); void registerPinotLLCRealtimeSegmentManager(PinotLLCRealtimeSegmentManager pinotLLCRealtimeSegmentManager); void updateTableConfig(TableConfig tableConfig); void setExistingTableConfig(TableConfig tableConfig); void updateMetadataConfigFor(String tableName, TableType type, TableCustomConfig newConfigs); void updateSegmentsValidationAndRetentionConfigFor(String tableName, TableType type,
      SegmentsValidationAndRetentionConfig newConfigs); void updateIndexingConfigFor(String tableName, TableType type, IndexingConfig newConfigs); void deleteOfflineTable(String tableName); void deleteRealtimeTable(String tableName); PinotResourceManagerResponse toggleTableState(String tableNameWithType, StateType stateType); void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl); void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
      @Nullable String crypter); @Nullable ZNRecord getSegmentMetadataZnRecord(String tableNameWithType, String segmentName); boolean updateZkMetadata(String offlineTableName, OfflineSegmentZKMetadata segmentMetadata,
      int expectedVersion); boolean updateZkMetadata(String offlineTableName, OfflineSegmentZKMetadata segmentMetadata); void refreshSegment(String offlineTableName, SegmentMetadata segmentMetadata,
      OfflineSegmentZKMetadata offlineSegmentZKMetadata, String downloadUrl, @Nullable String crypter); int reloadAllSegments(String tableNameWithType); int reloadSegment(String tableNameWithType, String segmentName); void sendSegmentRefreshMessage(String tableNameWithType, String segmentName, boolean refreshServerSegment,
      boolean refreshBrokerRouting); void toggleQueryQuotaStateForBroker(String brokerInstanceName, String state); Map<String, List<String>> getServerToSegmentsMap(String tableNameWithType); synchronized Map<String, String> getSegmentsCrcForTable(String tableNameWithType); String buildPathForSegmentMetadata(String tableNameWithType, String segmentName); boolean hasTable(String tableNameWithType); boolean hasOfflineTable(String tableName); boolean hasRealtimeTable(String tableName); @Nullable IdealState getTableIdealState(String tableNameWithType); @Nullable ExternalView getTableExternalView(String tableNameWithType); @Nullable TableConfig getTableConfig(String tableNameWithType); @Nullable TableConfig getOfflineTableConfig(String tableName); @Nullable TableConfig getRealtimeTableConfig(String tableName); @Nullable TableConfig getTableConfig(String tableName, TableType tableType); List<TableConfig> getTableConfigsForSchema(String schemaName); List<String> getServerInstancesForTable(String tableName, TableType tableType); List<String> getBrokerInstancesForTable(String tableName, TableType tableType); PinotResourceManagerResponse enableInstance(String instanceName); PinotResourceManagerResponse disableInstance(String instanceName); PinotResourceManagerResponse dropInstance(String instanceName); RebalanceResult rebalanceTable(String tableNameWithType, Configuration rebalanceConfig); boolean instanceExists(String instanceName); List<String> getOnlineUnTaggedBrokerInstanceList(); List<String> getOnlineUnTaggedServerInstanceList(); List<String> getOnlineInstanceList(); BiMap<String, String> getDataInstanceAdminEndpoints(Set<String> instances); List<String> getExistingTableNamesWithType(String tableName, @Nullable TableType tableType); String startReplaceSegments(String tableNameWithType, List<String> segmentsFrom, List<String> segmentsTo); void endReplaceSegments(String tableNameWithType, String segmentLineageEntryId); static final String APPEND; static final long EXTERNAL_VIEW_ONLINE_SEGMENTS_MAX_WAIT_MS; static final long EXTERNAL_VIEW_CHECK_INTERVAL_MS; }