@Test public void testSetUpNewTable() { testSetUpNewTable(2, 1, 4, true); testSetUpNewTable(2, 3, 0, false); testSetUpNewTable(2, 3, 4, false); testSetUpNewTable(2, 3, 8, false); testSetUpNewTable(8, 10, 4, false); }
public void setUpNewTable(TableConfig tableConfig, IdealState idealState) { Preconditions.checkState(!_isStopping, "Segment manager is stopping"); String realtimeTableName = tableConfig.getTableName(); LOGGER.info("Setting up new LLC table: {}", realtimeTableName); List<String> currentSegments = getAllSegments(realtimeTableName); for (String segmentName : currentSegments) { Preconditions.checkState(SegmentName.isHighLevelConsumerSegmentName(segmentName), "Cannot set up new LLC table: %s with existing non-HLC segment: %s", realtimeTableName, segmentName); } _flushThresholdUpdateManager.clearFlushThresholdUpdater(realtimeTableName); PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(), tableConfig.getIndexingConfig().getStreamConfigs()); InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig); int numPartitions = getNumPartitions(streamConfig); int numReplicas = getNumReplicas(tableConfig, instancePartitions); SegmentAssignment segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(_helixManager, tableConfig); Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap = Collections.singletonMap(InstancePartitionsType.CONSUMING, instancePartitions); long currentTimeMs = getCurrentTimeMs(); Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields(); for (int partitionId = 0; partitionId < numPartitions; partitionId++) { String segmentName = setupNewPartition(tableConfig, streamConfig, partitionId, currentTimeMs, instancePartitions, numPartitions, numReplicas); updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, segmentName, segmentAssignment, instancePartitionsMap); } setIdealState(realtimeTableName, idealState); }
PinotLLCRealtimeSegmentManager { public void setUpNewTable(TableConfig tableConfig, IdealState idealState) { Preconditions.checkState(!_isStopping, "Segment manager is stopping"); String realtimeTableName = tableConfig.getTableName(); LOGGER.info("Setting up new LLC table: {}", realtimeTableName); List<String> currentSegments = getAllSegments(realtimeTableName); for (String segmentName : currentSegments) { Preconditions.checkState(SegmentName.isHighLevelConsumerSegmentName(segmentName), "Cannot set up new LLC table: %s with existing non-HLC segment: %s", realtimeTableName, segmentName); } _flushThresholdUpdateManager.clearFlushThresholdUpdater(realtimeTableName); PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(), tableConfig.getIndexingConfig().getStreamConfigs()); InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig); int numPartitions = getNumPartitions(streamConfig); int numReplicas = getNumReplicas(tableConfig, instancePartitions); SegmentAssignment segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(_helixManager, tableConfig); Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap = Collections.singletonMap(InstancePartitionsType.CONSUMING, instancePartitions); long currentTimeMs = getCurrentTimeMs(); Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields(); for (int partitionId = 0; partitionId < numPartitions; partitionId++) { String segmentName = setupNewPartition(tableConfig, streamConfig, partitionId, currentTimeMs, instancePartitions, numPartitions, numReplicas); updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, segmentName, segmentAssignment, instancePartitionsMap); } setIdealState(realtimeTableName, idealState); } }
PinotLLCRealtimeSegmentManager { public void setUpNewTable(TableConfig tableConfig, IdealState idealState) { Preconditions.checkState(!_isStopping, "Segment manager is stopping"); String realtimeTableName = tableConfig.getTableName(); LOGGER.info("Setting up new LLC table: {}", realtimeTableName); List<String> currentSegments = getAllSegments(realtimeTableName); for (String segmentName : currentSegments) { Preconditions.checkState(SegmentName.isHighLevelConsumerSegmentName(segmentName), "Cannot set up new LLC table: %s with existing non-HLC segment: %s", realtimeTableName, segmentName); } _flushThresholdUpdateManager.clearFlushThresholdUpdater(realtimeTableName); PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(), tableConfig.getIndexingConfig().getStreamConfigs()); InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig); int numPartitions = getNumPartitions(streamConfig); int numReplicas = getNumReplicas(tableConfig, instancePartitions); SegmentAssignment segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(_helixManager, tableConfig); Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap = Collections.singletonMap(InstancePartitionsType.CONSUMING, instancePartitions); long currentTimeMs = getCurrentTimeMs(); Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields(); for (int partitionId = 0; partitionId < numPartitions; partitionId++) { String segmentName = setupNewPartition(tableConfig, streamConfig, partitionId, currentTimeMs, instancePartitions, numPartitions, numReplicas); updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, segmentName, segmentAssignment, instancePartitionsMap); } setIdealState(realtimeTableName, idealState); } PinotLLCRealtimeSegmentManager(PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics); }
PinotLLCRealtimeSegmentManager { public void setUpNewTable(TableConfig tableConfig, IdealState idealState) { Preconditions.checkState(!_isStopping, "Segment manager is stopping"); String realtimeTableName = tableConfig.getTableName(); LOGGER.info("Setting up new LLC table: {}", realtimeTableName); List<String> currentSegments = getAllSegments(realtimeTableName); for (String segmentName : currentSegments) { Preconditions.checkState(SegmentName.isHighLevelConsumerSegmentName(segmentName), "Cannot set up new LLC table: %s with existing non-HLC segment: %s", realtimeTableName, segmentName); } _flushThresholdUpdateManager.clearFlushThresholdUpdater(realtimeTableName); PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(), tableConfig.getIndexingConfig().getStreamConfigs()); InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig); int numPartitions = getNumPartitions(streamConfig); int numReplicas = getNumReplicas(tableConfig, instancePartitions); SegmentAssignment segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(_helixManager, tableConfig); Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap = Collections.singletonMap(InstancePartitionsType.CONSUMING, instancePartitions); long currentTimeMs = getCurrentTimeMs(); Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields(); for (int partitionId = 0; partitionId < numPartitions; partitionId++) { String segmentName = setupNewPartition(tableConfig, streamConfig, partitionId, currentTimeMs, instancePartitions, numPartitions, numReplicas); updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, segmentName, segmentAssignment, instancePartitionsMap); } setIdealState(realtimeTableName, idealState); } PinotLLCRealtimeSegmentManager(PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics); boolean getIsSplitCommitEnabled(); String getControllerVipUrl(); void stop(); void setUpNewTable(TableConfig tableConfig, IdealState idealState); void removeLLCSegments(IdealState idealState); @VisibleForTesting TableConfig getTableConfig(String realtimeTableName); void commitSegmentFile(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor); void commitSegmentMetadata(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor); long getCommitTimeoutMS(String realtimeTableName); void segmentStoppedConsuming(LLCSegmentName llcSegmentName, String instanceName); void ensureAllPartitionsConsuming(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig); }
PinotLLCRealtimeSegmentManager { public void setUpNewTable(TableConfig tableConfig, IdealState idealState) { Preconditions.checkState(!_isStopping, "Segment manager is stopping"); String realtimeTableName = tableConfig.getTableName(); LOGGER.info("Setting up new LLC table: {}", realtimeTableName); List<String> currentSegments = getAllSegments(realtimeTableName); for (String segmentName : currentSegments) { Preconditions.checkState(SegmentName.isHighLevelConsumerSegmentName(segmentName), "Cannot set up new LLC table: %s with existing non-HLC segment: %s", realtimeTableName, segmentName); } _flushThresholdUpdateManager.clearFlushThresholdUpdater(realtimeTableName); PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(), tableConfig.getIndexingConfig().getStreamConfigs()); InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig); int numPartitions = getNumPartitions(streamConfig); int numReplicas = getNumReplicas(tableConfig, instancePartitions); SegmentAssignment segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(_helixManager, tableConfig); Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap = Collections.singletonMap(InstancePartitionsType.CONSUMING, instancePartitions); long currentTimeMs = getCurrentTimeMs(); Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields(); for (int partitionId = 0; partitionId < numPartitions; partitionId++) { String segmentName = setupNewPartition(tableConfig, streamConfig, partitionId, currentTimeMs, instancePartitions, numPartitions, numReplicas); updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, segmentName, segmentAssignment, instancePartitionsMap); } setIdealState(realtimeTableName, idealState); } PinotLLCRealtimeSegmentManager(PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics); boolean getIsSplitCommitEnabled(); String getControllerVipUrl(); void stop(); void setUpNewTable(TableConfig tableConfig, IdealState idealState); void removeLLCSegments(IdealState idealState); @VisibleForTesting TableConfig getTableConfig(String realtimeTableName); void commitSegmentFile(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor); void commitSegmentMetadata(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor); long getCommitTimeoutMS(String realtimeTableName); void segmentStoppedConsuming(LLCSegmentName llcSegmentName, String instanceName); void ensureAllPartitionsConsuming(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig);  }