deepthi912 commented on code in PR #20106:
URL: https://github.com/apache/pinot/pull/20106#discussion_r4446093097
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java:
##########
@@ -116,14 +121,65 @@ public void acquire(LLCSegmentName llcSegmentName)
}
}
+ /// Waits for required history before acquiring the consumer permit. Local
permit releases wake all waiters;
+ /// Helix assignment and state changes still use the periodic check.
Readiness can perform ZooKeeper reads, so it
+ /// must run outside the monitor. The version prevents losing a release
between the readiness check and the wait.
+ /// @return whether readiness passed, or false if consumption was stopped
+ boolean waitUntilReadyToConsume(int partitionGroupId, BooleanSupplier
isReadyToConsume, BooleanSupplier shouldStop)
Review Comment:
With `enforceConsumptionInOrder` on, a dedup/partial-upsert consumer now
goes through two "is my history loaded" checks back to back: this readiness
wait, then `acquire()` → `waitForPreviousSegment()`. For these tables the
readiness check is strictly stronger (every earlier ONLINE segment of the
partition, not just N-1), so `waitForPreviousSegment()` adds nothing once this
returns true.
The two also scan the same IdealState map with nearly the same filter:
`getPreviousSegmentSequenceNumberFromIdealState()` (ONLINE on this instance,
same partition, lower sequence) and `TableStateUtils.isRequiredHistory()` (the
same, plus uploaded segments). But they use different signals for "loaded":
`_maxSequenceNumberRegistered` vs. CurrentState / `isMetadataReady()` /
`hasReleasedConsumerSemaphore()`.
Could the history check live in `ConsumerCoordinator` instead, e.g. as the
`waitForPreviousSegment()` implementation for dedup/partial-upsert tables? That
would give:
- one IdealState scan and one definition of "previous segment loaded"
- one wait loop instead of `_readinessMonitor` alongside `_lock`/`_condition`
- `register()` (called at the end of `addOnlineSegment`) waking the waiters,
so history that arrives through OFFLINE→ONLINE during a rebalance is picked up
without waiting for the 5s poll
--
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]
