Jackie-Jiang commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2874839268
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -400,7 +400,15 @@ public void setUpNewTable(TableConfig tableConfig,
IdealState idealState,
List<StreamConfig> streamConfigs =
IngestionConfigUtils.getStreamConfigs(tableConfig);
streamConfigs.forEach(_flushThresholdUpdateManager::clearFlushThresholdUpdater);
InstancePartitions instancePartitions =
getConsumingInstancePartitions(tableConfig);
- int numPartitionGroups = consumeMeta.size();
+ // For subset-partition tables the instance map was built with the full
Kafka topic partition
+ // count (set by ImplicitRealtimeTablePartitionSelector), so
getNumPartitions() exceeds the
+ // number of actively consumed partitions. Use it so that
RealtimeSegmentAssignment correctly
+ // resolves non-contiguous partition IDs via partitionId % numPartitions.
+ // For standard tables the instance map either has a single slot
(non-partition-based
+ // assignment) or exactly as many slots as partitions, so
consumeMeta.size() is the right value.
+ int numPartitionGroups = instancePartitions.getNumPartitions() >
consumeMeta.size()
Review Comment:
This is not correct because instance partitions might not match kafka stream
(e.g. when instance assignment is not explicitly configured). We should read
the total partitions from the stream, and pass it in.
Same for other places
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
