@Test public void testRebuildBrokerResourceFromHelixTags() throws Exception { Tenant brokerTenant = new Tenant(TenantRole.BROKER, BROKER_TENANT_NAME, 3, 0, 0); PinotResourceManagerResponse response = _helixResourceManager.createBrokerTenant(brokerTenant); Assert.assertTrue(response.isSuccessful()); TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setNumReplicas(3) .setBrokerTenant(BROKER_TENANT_NAME).setServerTenant(SERVER_TENANT_NAME).build(); _helixResourceManager.addTable(tableConfig); IdealState idealState = _helixResourceManager.getHelixAdmin() .getResourceIdealState(getHelixClusterName(), CommonConstants.Helix.BROKER_RESOURCE_INSTANCE); untagBrokers(); Assert.assertEquals(_helixResourceManager.getOnlineUnTaggedBrokerInstanceList().size(), NUM_INSTANCES); response = _helixResourceManager.rebuildBrokerResourceFromHelixTags(OFFLINE_TABLE_NAME); Assert.assertTrue(response.isSuccessful()); idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(), CommonConstants.Helix.BROKER_RESOURCE_INSTANCE); Assert.assertEquals(idealState.getInstanceStateMap(OFFLINE_TABLE_NAME).size(), 0); brokerTenant = new Tenant(TenantRole.BROKER, BROKER_TENANT_NAME, 5, 0, 0); response = _helixResourceManager.createBrokerTenant(brokerTenant); Assert.assertTrue(response.isSuccessful()); response = _helixResourceManager.rebuildBrokerResourceFromHelixTags(OFFLINE_TABLE_NAME); Assert.assertTrue(response.isSuccessful()); idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(), CommonConstants.Helix.BROKER_RESOURCE_INSTANCE); Assert.assertEquals(idealState.getInstanceStateMap(OFFLINE_TABLE_NAME).size(), 5); _helixResourceManager.deleteOfflineTable(TABLE_NAME); untagBrokers(); Assert.assertEquals(_helixResourceManager.getOnlineUnTaggedBrokerInstanceList().size(), NUM_INSTANCES); }
public PinotResourceManagerResponse rebuildBrokerResourceFromHelixTags(String tableNameWithType) throws Exception { TableConfig tableConfig; try { tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, tableNameWithType); } catch (Exception e) { LOGGER.warn("Caught exception while getting table config for table {}", tableNameWithType, e); throw new InvalidTableConfigException( "Failed to fetch broker tag for table " + tableNameWithType + " due to exception: " + e.getMessage()); } if (tableConfig == null) { LOGGER.warn("Table " + tableNameWithType + " does not exist"); throw new InvalidConfigException( "Invalid table configuration for table " + tableNameWithType + ". Table does not exist"); } return rebuildBrokerResource(tableNameWithType, getAllInstancesForBrokerTenant(tableConfig.getTenantConfig().getBroker())); }
PinotHelixResourceManager { public PinotResourceManagerResponse rebuildBrokerResourceFromHelixTags(String tableNameWithType) throws Exception { TableConfig tableConfig; try { tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, tableNameWithType); } catch (Exception e) { LOGGER.warn("Caught exception while getting table config for table {}", tableNameWithType, e); throw new InvalidTableConfigException( "Failed to fetch broker tag for table " + tableNameWithType + " due to exception: " + e.getMessage()); } if (tableConfig == null) { LOGGER.warn("Table " + tableNameWithType + " does not exist"); throw new InvalidConfigException( "Invalid table configuration for table " + tableNameWithType + ". Table does not exist"); } return rebuildBrokerResource(tableNameWithType, getAllInstancesForBrokerTenant(tableConfig.getTenantConfig().getBroker())); } }
PinotHelixResourceManager { public PinotResourceManagerResponse rebuildBrokerResourceFromHelixTags(String tableNameWithType) throws Exception { TableConfig tableConfig; try { tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, tableNameWithType); } catch (Exception e) { LOGGER.warn("Caught exception while getting table config for table {}", tableNameWithType, e); throw new InvalidTableConfigException( "Failed to fetch broker tag for table " + tableNameWithType + " due to exception: " + e.getMessage()); } if (tableConfig == null) { LOGGER.warn("Table " + tableNameWithType + " does not exist"); throw new InvalidConfigException( "Invalid table configuration for table " + tableNameWithType + ". Table does not exist"); } return rebuildBrokerResource(tableNameWithType, getAllInstancesForBrokerTenant(tableConfig.getTenantConfig().getBroker())); } PinotHelixResourceManager(String zkURL, String helixClusterName, @Nullable String dataDir,
      long externalViewOnlineToOfflineTimeoutMillis, boolean isSingleTenantCluster, boolean enableBatchMessageMode,
      boolean allowHLCTables);  PinotHelixResourceManager(ControllerConf controllerConf); }
PinotHelixResourceManager { public PinotResourceManagerResponse rebuildBrokerResourceFromHelixTags(String tableNameWithType) throws Exception { TableConfig tableConfig; try { tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, tableNameWithType); } catch (Exception e) { LOGGER.warn("Caught exception while getting table config for table {}", tableNameWithType, e); throw new InvalidTableConfigException( "Failed to fetch broker tag for table " + tableNameWithType + " due to exception: " + e.getMessage()); } if (tableConfig == null) { LOGGER.warn("Table " + tableNameWithType + " does not exist"); throw new InvalidConfigException( "Invalid table configuration for table " + tableNameWithType + ". Table does not exist"); } return rebuildBrokerResource(tableNameWithType, getAllInstancesForBrokerTenant(tableConfig.getTenantConfig().getBroker())); } 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 PinotResourceManagerResponse rebuildBrokerResourceFromHelixTags(String tableNameWithType) throws Exception { TableConfig tableConfig; try { tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, tableNameWithType); } catch (Exception e) { LOGGER.warn("Caught exception while getting table config for table {}", tableNameWithType, e); throw new InvalidTableConfigException( "Failed to fetch broker tag for table " + tableNameWithType + " due to exception: " + e.getMessage()); } if (tableConfig == null) { LOGGER.warn("Table " + tableNameWithType + " does not exist"); throw new InvalidConfigException( "Invalid table configuration for table " + tableNameWithType + ". Table does not exist"); } return rebuildBrokerResource(tableNameWithType, getAllInstancesForBrokerTenant(tableConfig.getTenantConfig().getBroker())); } 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; }