Jackie-Jiang commented on code in PR #20106:
URL: https://github.com/apache/pinot/pull/20106#discussion_r4448374572
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/tablestate/TableStateUtils.java:
##########
@@ -82,12 +84,56 @@ public static List<String>
getSegmentsInGivenStateForThisInstance(HelixManager h
public static boolean isAllSegmentsLoaded(HelixManager helixManager, String
tableNameWithType) {
List<String> onlineSegments =
getSegmentsInGivenStateForThisInstance(helixManager,
tableNameWithType, SegmentStateModel.ONLINE);
+ return areSegmentsLoaded(helixManager, tableNameWithType, onlineSegments);
+ }
+
+ /// Checks the current assignment before a consumer starts. Earlier ONLINE
segments of the same partition must be
+ /// loaded. Non-LLC ONLINE segments are conservatively included because they
may hold base rows for the partition.
+ /// The consumer itself and its successors are excluded so that they do not
wait for their own completion.
+ ///
+ /// @param isLocallyReady identifies segments whose metadata is already
ready locally despite a pending ONLINE
+ /// transition, such as a predecessor that finished
consumption and released its permit
+ public static boolean isReadyToConsume(HelixManager helixManager, String
tableNameWithType, LLCSegmentName consumer,
+ Predicate<String> isLocallyReady) {
+ HelixDataAccessor dataAccessor = helixManager.getHelixDataAccessor();
+ IdealState idealState =
dataAccessor.getProperty(dataAccessor.keyBuilder().idealStates(tableNameWithType));
+ if (idealState == null) {
+ LOGGER.warn("Failed to find ideal state for table: {}",
tableNameWithType);
+ return false;
+ }
+ String instanceName = helixManager.getInstanceName();
+ Map<String, String> consumerAssignment =
idealState.getInstanceStateMap(consumer.getSegmentName());
+ String consumerState = consumerAssignment != null ?
consumerAssignment.get(instanceName) : null;
+ if (!SegmentStateModel.CONSUMING.equals(consumerState) &&
!SegmentStateModel.ONLINE.equals(consumerState)) {
+ return false;
+ }
+
+ List<String> requiredSegments = new ArrayList<>();
+ for (Map.Entry<String, Map<String, String>> entry :
idealState.getRecord().getMapFields().entrySet()) {
+ if
(!SegmentStateModel.ONLINE.equals(entry.getValue().get(instanceName))) {
+ continue;
+ }
+ String segmentName = entry.getKey();
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
+ if (llcSegmentName != null && (llcSegmentName.getPartitionGroupId() !=
consumer.getPartitionGroupId()
+ || llcSegmentName.getSequenceNumber() >=
consumer.getSequenceNumber())) {
+ continue;
+ }
+ if (!isLocallyReady.test(segmentName)) {
+ requiredSegments.add(segmentName);
Review Comment:
Waiting for unavailable required history, including ERROR segments, is
intentional for these modes: consuming without that metadata can produce
incorrect partial-upsert, dedup, or out-of-order decisions. The PR now exposes
a per-partition `CONSUMERS_WAITING_FOR_READINESS` gauge and logs the first
unavailable history segment and its state on fresh checks. The gauge counts
overlapping waiters and is removed when the last waiter exits. Uploaded
segments with a known different partition are excluded; unparseable partition
names remain conservatively included. The PR description documents that
unavailable required history blocks ingestion.
--
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]
