shounakmk219 commented on code in PR #19170:
URL: https://github.com/apache/pinot/pull/19170#discussion_r3863510246
##########
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:
this can happen if a partition looses its consuming segment between the
snapshot and the lock or retries on updater. Not falling back to
fetchPartitionGroupIdToSmallestOffset as this PR is trying to pull out all the
IO ops out of the IS lock, it should be picked in the next RVM run anyways.
##########
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:
yes it should have all the partitionId entries fetched from upstream
--
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]
