noob-se7en commented on code in PR #19170:
URL: https://github.com/apache/pinot/pull/19170#discussion_r3863140851
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1861,16 +1936,21 @@ IdealState ensureAllPartitionsConsuming(TableConfig
tableConfig, List<StreamConf
continue;
}
- // Smallest offset is fetched from stream once and cached in
partitionIdToSmallestOffset.
+ // The smallest offsets were not fetched this cycle (the lock-free
snapshot gate saw no partition needing a
+ // new CONSUMING segment) but this partition needs one now - it
started needing repair after the snapshot.
+ // Defer to the next validation run, which will fetch the true
stream-smallest for it. Start offsets are not
+ // the stream-smallest here, so they must not be substituted.
if (partitionIdToSmallestOffset == null) {
- partitionIdToSmallestOffset =
- fetchPartitionGroupIdToSmallestOffset(streamConfigs,
idealState, latestSegmentZKMetadataMap);
+ LOGGER.info("Smallest stream offsets not fetched this cycle;
deferring repair of partition: {} of table: "
+ + "{} to the next run", partitionId, realtimeTableName);
+ continue;
}
-
- // Do not create new CONSUMING segment when the stream partition has
reached end of life.
- if (!partitionIdToSmallestOffset.containsKey(partitionId)) {
- LOGGER.info("PartitionGroup: {} has reached end of life. Skipping
creation of new segment {}", partitionId,
- latestSegmentName);
+ // Do not create a new CONSUMING segment when the partition has no
smallest stream offset (it has reached
+ // end of life).
+ StreamPartitionMsgOffset smallestStreamOffset =
partitionIdToSmallestOffset.get(partitionId);
Review Comment:
Is it guaranteed `partitionIdToSmallestOffset` will have partitionId?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1861,16 +1936,21 @@ IdealState ensureAllPartitionsConsuming(TableConfig
tableConfig, List<StreamConf
continue;
}
- // Smallest offset is fetched from stream once and cached in
partitionIdToSmallestOffset.
+ // The smallest offsets were not fetched this cycle (the lock-free
snapshot gate saw no partition needing a
+ // new CONSUMING segment) but this partition needs one now - it
started needing repair after the snapshot.
+ // Defer to the next validation run, which will fetch the true
stream-smallest for it. Start offsets are not
+ // the stream-smallest here, so they must not be substituted.
if (partitionIdToSmallestOffset == null) {
- partitionIdToSmallestOffset =
- fetchPartitionGroupIdToSmallestOffset(streamConfigs,
idealState, latestSegmentZKMetadataMap);
+ LOGGER.info("Smallest stream offsets not fetched this cycle;
deferring repair of partition: {} of table: "
Review Comment:
can this happen? If yes, then why not fetchPartitionGroupIdToSmallestOffset
here as fallback?
--
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]
