Copilot commented on code in PR #18784:
URL: https://github.com/apache/pinot/pull/18784#discussion_r3431722189
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -484,6 +485,16 @@ protected boolean consumeLoop()
_segmentLogger.info("Starting consumption loop start offset {},
finalOffset {}", _currentOffset, _finalOffset);
while (!_shouldStop && !endCriteriaReached()) {
+ if (_state == State.INITIAL_CONSUMING &&
_serverIngestionOomProtectionManager != null) {
+ boolean waitedForOomProtection =
+ _serverIngestionOomProtectionManager.waitIfProtectionNeeded(() ->
_shouldStop || endCriteriaReached());
+ if (_shouldStop || endCriteriaReached()) {
+ break;
+ }
+ if (waitedForOomProtection) {
+ _idleTimer.markStreamCreated();
+ }
+ }
Review Comment:
The stopCondition passed into waitIfProtectionNeeded() calls
endCriteriaReached(), which mutates state (e.g., it can extend _consumeEndTime
when no messages have been fetched yet). While throttled, this can change
segment end behavior just because OOM protection is active (e.g., hitting the
initial time limit during a throttle wait can trigger the "No events came in,
extending time" path). Consider using a side-effect-free stop predicate that
checks the relevant fields directly, and leave the single authoritative
endCriteriaReached() call to the main loop.
--
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]
