deepthi912 commented on code in PR #20106:
URL: https://github.com/apache/pinot/pull/20106#discussion_r4533865912
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/tablestate/TableStateUtils.java:
##########
@@ -82,30 +87,130 @@ public static List<String>
getSegmentsInGivenStateForThisInstance(HelixManager h
public static boolean isAllSegmentsLoaded(HelixManager helixManager, String
tableNameWithType) {
List<String> onlineSegments =
getSegmentsInGivenStateForThisInstance(helixManager,
tableNameWithType, SegmentStateModel.ONLINE);
- if (onlineSegments.isEmpty()) {
- LOGGER.info("No ONLINE segment found for table: {}", tableNameWithType);
- return true;
- }
+ return areSegmentsLoaded(helixManager, tableNameWithType, onlineSegments);
+ }
- // Check if ideal state and current state matches for all segments
assigned to the current instance
+ /// Checks the current assignment before a consumer starts. Earlier ONLINE
segments of the same partition must be
+ /// loaded, as must uploaded segments from that partition. Segments whose
partition cannot be inferred from their
+ /// name are conservatively included. The consumer and its successors are
excluded to avoid waiting on themselves.
+ ///
+ /// @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();
- PropertyKey.Builder keyBuilder = dataAccessor.keyBuilder();
- String instanceName = helixManager.getInstanceName();
- LiveInstance liveInstance =
dataAccessor.getProperty(keyBuilder.liveInstance(instanceName));
- if (liveInstance == null) {
- LOGGER.warn("Failed to find live instance for instance: {}",
instanceName);
+ IdealState idealState =
dataAccessor.getProperty(dataAccessor.keyBuilder().idealStates(tableNameWithType));
+ if (idealState == null) {
+ LOGGER.warn("Failed to find ideal state for table: {}",
tableNameWithType);
return false;
}
- String sessionId = liveInstance.getEphemeralOwner();
- CurrentState currentState =
- dataAccessor.getProperty(keyBuilder.currentState(instanceName,
sessionId, tableNameWithType));
- if (currentState == null) {
- LOGGER.warn("Failed to find current state for instance: {}, sessionId:
{}, table: {}", instanceName, sessionId,
- tableNameWithType);
+ 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();
+ if (isRequiredHistory(segmentName, consumer) &&
!isLocallyReady.test(segmentName)) {
+ requiredSegments.add(segmentName);
+ }
+ }
+ return areSegmentsLoaded(helixManager, tableNameWithType,
requiredSegments);
+ }
+
+ /// Reads the table assignment and this live instance's current state once
for evaluation by multiple consumers.
+ /// The records are copied so subsequent Helix updates cannot mutate the
returned snapshot. A cached snapshot must
+ /// only reject readiness; callers must perform a fresh check before
authorizing consumption from a cached result.
+ public static ReadinessSnapshot getReadinessSnapshot(HelixManager
helixManager, String tableNameWithType) {
+ 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 new ReadinessSnapshot(Map.of(), Map.of());
+ }
+ String instanceName = helixManager.getInstanceName();
+ Map<String, String> idealStates = new HashMap<>();
+ for (Map.Entry<String, Map<String, String>> entry :
idealState.getRecord().getMapFields().entrySet()) {
+ String state = entry.getValue().get(instanceName);
+ if (state != null) {
+ idealStates.put(entry.getKey(), state);
+ }
+ }
+ return new ReadinessSnapshot(idealStates, getCurrentStates(helixManager,
tableNameWithType));
+ }
+
+ /// An immutable view of this instance's assignment and current state, safe
to share between consumer threads.
+ /// Evaluation performs no ZooKeeper reads and reevaluates the supplied
local-readiness predicate for each required
+ /// segment. The predicate must itself be safe for its caller's threads; its
results are never retained here.
+ public static final class ReadinessSnapshot {
+ private final Map<String, String> _idealStates;
+ private final Map<String, String> _currentStates;
+
+ private ReadinessSnapshot(Map<String, String> idealStates, Map<String,
String> currentStates) {
+ _idealStates = Collections.unmodifiableMap(new HashMap<>(idealStates));
+ _currentStates = Collections.unmodifiableMap(new
HashMap<>(currentStates));
+ }
+
+ public boolean isReadyToConsume(LLCSegmentName consumer, Predicate<String>
isLocallyReady) {
+ return isReadyToConsume(consumer, isLocallyReady, false);
+ }
+
+ /// Optionally logs the first unavailable history segment. Enable logging
for a freshly fetched snapshot, keeping
+ /// repeated rejection checks against a cached snapshot quiet.
+ public boolean isReadyToConsume(LLCSegmentName consumer, Predicate<String>
isLocallyReady,
+ boolean logUnavailableSegments) {
+ String consumerState = _idealStates.get(consumer.getSegmentName());
+ if (!SegmentStateModel.CONSUMING.equals(consumerState) &&
!SegmentStateModel.ONLINE.equals(consumerState)) {
+ return false;
+ }
+ for (Map.Entry<String, String> entry : _idealStates.entrySet()) {
+ String segmentName = entry.getKey();
+ if (SegmentStateModel.ONLINE.equals(entry.getValue()) &&
isRequiredHistory(segmentName, consumer)
+ && !isLocallyReady.test(segmentName)) {
+ String actualState = _currentStates.get(segmentName);
+ if (!SegmentStateModel.ONLINE.equals(actualState)) {
+ if (logUnavailableSegments) {
+ if (SegmentStateModel.ERROR.equals(actualState)) {
+ LOGGER.error("Consumer: {} is waiting for history segment: {}
in ERROR state, expected: {}",
+ consumer.getSegmentName(), segmentName,
SegmentStateModel.ONLINE);
+ } else {
+ LOGGER.info("Consumer: {} is waiting for history segment: {}
in state: {}, expected: {}",
+ consumer.getSegmentName(), segmentName, actualState,
SegmentStateModel.ONLINE);
+ }
+ }
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+ }
+
+ private static boolean isRequiredHistory(String segmentName, LLCSegmentName
consumer) {
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
+ if (llcSegmentName != null) {
+ return llcSegmentName.getPartitionGroupId() ==
consumer.getPartitionGroupId()
+ && llcSegmentName.getSequenceNumber() < consumer.getSequenceNumber();
+ }
+ Integer partitionId =
SegmentUtils.getPartitionIdFromSegmentName(segmentName);
+ return partitionId == null || partitionId ==
consumer.getPartitionGroupId();
Review Comment:
A segment that is neither LLC nor `UploadedRealtimeSegmentName` (e.g. a
backfill pushed with a default name) gets `null` here and becomes required
history for every partition, so one such segment in ERROR stops the whole
table. But `handleUpsert` / `handleDedup` already place it in one partition via
`SegmentUtils.getSegmentPartitionId(...)` (name, then ZK metadata). Could the
gate use the same lookup (resolved when building the snapshot, cached per
segment), keeping `null → required` only when that also fails?
--
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]
