[GitHub] [pinot] sajjad-moradi commented on a change in pull request #7066: Add support for lowest partition id in case partition id 0 is not available

[GitBox <gi...@apache.org>]

sajjad-moradi commented on a change in pull request #7066:
URL: https://github.com/apache/pinot/pull/7066#discussion_r727432780



##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentSizeBasedFlushThresholdUpdater.java
##########
@@ -103,8 +107,15 @@ public synchronized void updateFlushThreshold(PartitionLevelStreamConfig streamC
     // less same characteristics at any one point in time).

Review comment:
       Could you please update the first comment line to reflect the changes this PR introduces?

##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentSizeBasedFlushThresholdUpdater.java
##########
@@ -102,8 +106,14 @@ public synchronized void updateFlushThreshold(PartitionLevelStreamConfig streamC
     // less same characteristics at any one point in time).
     // However, when we start a new table or change controller mastership, we can have any partition completing first.
     // It is best to learn the ratio as quickly as we can, so we allow any partition to supply the value.
-    // FIXME: The stream may not have partition "0"
-    if (new LLCSegmentName(newSegmentName).getPartitionGroupId() == 0 || _latestSegmentRowsToSizeRatio == 0) {
+
+    // Partition group id 0 might not be available always. We take the smallest available partition id in that case to update the threshold
+    int smallestAvailablePartitionGroupId =
+        partitionGroupMetadataList.stream().min(Comparator.comparingInt(PartitionGroupMetadata::getPartitionGroupId))
+            .map(PartitionGroupMetadata::getPartitionGroupId).orElseGet(() -> 0);

Review comment:
       `updateFlushThreshod` method gets called only in `PinotLLCRealtimeSegmentManager.createNewSegmentZKMetadata` which is for creating ZK metadata for new segments. If there's no partition available, it means there's no new segments and updateFlushThreshold doesn't get called. So, we can throw illegal state exception for the orElse part.
   Also, min calculation can be simplified to:
   ```java
   partitionGroupMetadataList.stream()
     .mapToInt(PartitionGroupMetadata::getPartitionGroupId)
     .min()
     .orElseThrow(() -> new IllegalStateException("..."));
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org