npawar commented on a change in pull request #6667:
URL: https://github.com/apache/incubator-pinot/pull/6667#discussion_r595600886
##########
File path:
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
##########
@@ -448,14 +494,39 @@ private void commitSegmentMetadataInternal(String
realtimeTableName,
// Refresh the Broker routing to reflect the changes in the segment ZK
metadata
_helixResourceManager.sendSegmentRefreshMessage(realtimeTableName,
committingSegmentName, false, true);
- // Step-2
+ // Get current partition groups - this gives current state of latest
segments for each partition
+ // E.g. [A - DONE], [B - IN_PROGRESS], [C - IN_PROGRESS]
+ PartitionLevelStreamConfig streamConfig = new
PartitionLevelStreamConfig(tableConfig.getTableName(),
+ IngestionConfigUtils.getStreamConfigMap(tableConfig));
+ List<PartitionGroupMetadata> currentPartitionGroupMetadataList =
+ getCurrentPartitionGroupMetadataList(idealState, streamConfig);
+
+ // Fetches new partition groups, given current partition groups metadata.
+ // Assume stream has partitions A, B, C, all still consuming. Result will
be A, B, C
+ // Assume A was split into D, E, but messages of A are yet to be consumed,
result will be A, B, C
Review comment:
The comments are kinda irrelevant out here, as those are kinesis
specific examples. Removed them.
In general, the controller will get the currentPartitionGroupMetadata using
the ideal state and segment zk metadata of latest segment for each partition
group.
The Stream impl will look at the partitions in the stream and get new
partition groups.
Take the example of Kafka, if we began with 3 partitions, and partition 0
was committing, we would have
Current partition group metadata:
0 - seqNum: 0, startOffset: 100, endOffset: 220, status: DONE
1 - seqNum: 0, startOffset: 110, endOffset: -, status: IN_PROGRESS
2 - seqNum: 0, startOffset: 100, endOffset: -, status: IN_PROGRESS
And new partition group info: [0, 1, 2]
Lets assume partition 3,4 had come up, before the commit call for partition
0 was reached. Then, the new partition group info would be: [0, 1, 2, 3, 4]
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
