Jackie-Jiang commented on code in PR #15404:
URL: https://github.com/apache/pinot/pull/15404#discussion_r2019655819
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java:
##########
@@ -115,177 +109,111 @@ Semaphore getSemaphore() {
return _semaphore;
}
- public void trackSegment(LLCSegmentName llcSegmentName) {
+ public void register(LLCSegmentName llcSegmentName) {
_lock.lock();
try {
- if (!_alwaysRelyOnIdealState) {
- _maxSegmentSeqNumRegistered = Math.max(_maxSegmentSeqNumRegistered,
llcSegmentName.getSequenceNumber());
+ int sequenceNumber = llcSegmentName.getSequenceNumber();
+ if (sequenceNumber > _maxSequenceNumberRegistered) {
+ _maxSequenceNumberRegistered = sequenceNumber;
+ // notify all helix threads waiting for their offline -> consuming
segment's prev segment to be loaded
+ _condition.signalAll();
}
- // notify all helix threads waiting for their offline -> consuming
segment's prev segment to be loaded
- _condition.signalAll();
} finally {
_lock.unlock();
}
}
- private void waitForPrevSegment(LLCSegmentName currSegment)
- throws InterruptedException {
-
- if (_alwaysRelyOnIdealState || !_firstTransitionProcessed.get()) {
- // if _alwaysRelyOnIdealState or no offline -> consuming transition has
been processed, it means rely on
- // ideal state to fetch previous segment.
- awaitForPreviousSegmentFromIdealState(currSegment);
-
- // the first transition will always be prone to error, consider edge
case where segment previous to current
- // helix transition's segment was deleted and this server came alive
after successful deletion. the prev
- // segment will not exist, hence first transition is handled using
isFirstTransitionSuccessful.
- _firstTransitionProcessed.set(true);
- return;
- }
-
- // rely on _maxSegmentSeqNumRegistered watermark for previous segment.
- if (awaitForPreviousSegmentSequenceNumber(currSegment, WAIT_INTERVAL_MS)) {
- return;
- }
-
- // tried using prevSegSeqNumber watermark, but could not acquire the
previous segment.
- // fallback to acquire prev segment from ideal state.
- awaitForPreviousSegmentFromIdealState(currSegment);
- }
-
- private void awaitForPreviousSegmentFromIdealState(LLCSegmentName
currSegment)
+ private void waitForPreviousSegment(LLCSegmentName currentSegment)
throws InterruptedException {
- String previousSegment = getPreviousSegmentFromIdealState(currSegment);
- if (previousSegment == null) {
- // previous segment can only be null if either all the previous segments
are deleted or this is the starting
- // sequence segment of the partition Group.
- return;
- }
-
- SegmentDataManager segmentDataManager =
_realtimeTableDataManager.acquireSegment(previousSegment);
- try {
- long startTimeMs = System.currentTimeMillis();
- _lock.lock();
- try {
- while (segmentDataManager == null) {
- // if segmentDataManager == null, it means segment is not loaded in
the server.
- // wait until it's loaded.
- if (!_condition.await(WAIT_INTERVAL_MS, TimeUnit.MILLISECONDS)) {
- LOGGER.warn("Semaphore access denied to segment: {}. Waited on
previous segment: {} for: {}ms.",
- currSegment.getSegmentName(), previousSegment,
System.currentTimeMillis() - startTimeMs);
-
- // waited until timeout, fetch previous segment again from ideal
state as previous segment might be
- // changed in ideal state.
- previousSegment = getPreviousSegmentFromIdealState(currSegment);
- if (previousSegment == null) {
- return;
- }
- }
- segmentDataManager =
_realtimeTableDataManager.acquireSegment(previousSegment);
- }
- } finally {
- _lock.unlock();
- }
- } finally {
- if (segmentDataManager != null) {
- _realtimeTableDataManager.releaseSegment(segmentDataManager);
+ if (!_firstTransitionProcessed.get() ||
_useIdealStateToCalculatePreviousSegment) {
+ if (_maxSequenceNumberRegistered < currentSegment.getSequenceNumber() -
1) {
Review Comment:
Do a quick check to reduce overhead
--
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: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
