You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "Jackie-Jiang (via GitHub)" <gi...@apache.org> on 2025/03/29 01:34:37 UTC
[PR] Fix starvation in consumer lock [pinot]
Jackie-Jiang opened a new pull request, #15404:
URL: https://github.com/apache/pinot/pull/15404
Currently the consumer lock is acquired during the `OFFLINE -> CONSUMING` state transition. If the lock cannot be acquired, helix thread will be blocked, and end up causing starvation. With #15261 introducing dependency among segments, the chance of starvation increases.
This PR moves the lock acquisition into the consumer thread so that it won't block the helix state transition.
This PR also modifies the consumer coordinator logic:
- Only use ideal state to calculate the previous sequence number, always use watermark to determine whether previous segment is registered
- Fix a bug where deleted segment is not properly handled
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix starvation in consumer lock [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #15404:
URL: https://github.com/apache/pinot/pull/15404#discussion_r2019674182
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -557,22 +565,10 @@ private void doAddConsumingSegment(String segmentName)
PartitionDedupMetadataManager partitionDedupMetadataManager =
_tableDedupMetadataManager != null ? _tableDedupMetadataManager.getOrCreatePartitionManager(partitionGroupId)
: null;
- RealtimeSegmentDataManager realtimeSegmentDataManager;
- try {
- realtimeSegmentDataManager =
- createRealtimeSegmentDataManager(zkMetadata, tableConfig, indexLoadingConfig, schema, llcSegmentName,
- consumerCoordinator, partitionUpsertMetadataManager, partitionDedupMetadataManager,
- _isTableReadyToConsumeData);
- } catch (SegmentAlreadyConsumedException e) {
Review Comment:
Lock is applied when starting the consumer thread, so this part won't throw this exception any more. This exception is also changed to be checked exception so that it won't be missed
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix starvation in consumer lock [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #15404:
URL: https://github.com/apache/pinot/pull/15404#discussion_r2019656629
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java:
##########
@@ -299,26 +227,36 @@ AtomicBoolean getFirstTransitionProcessed() {
return _firstTransitionProcessed;
}
- // this should not be used outside of tests.
@VisibleForTesting
- int getMaxSegmentSeqNumLoaded() {
- return _maxSegmentSeqNumRegistered;
+ int getMaxSequenceNumberRegistered() {
+ return _maxSequenceNumberRegistered;
}
@VisibleForTesting
- boolean isSegmentAlreadyConsumed(String currSegmentName) {
- SegmentZKMetadata segmentZKMetadata = _realtimeTableDataManager.fetchZKMetadata(currSegmentName);
+ void checkSegmentZKMetadataStatus(String segmentName)
+ throws ShouldNotConsumeException {
+ SegmentZKMetadata segmentZKMetadata = _realtimeTableDataManager.fetchZKMetadataNullable(segmentName);
Review Comment:
This is a bug fix. Existing code directly throw exception when segment is deleted
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix starvation in consumer lock [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on code in PR #15404:
URL: https://github.com/apache/pinot/pull/15404#discussion_r2019977505
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -474,23 +473,26 @@ protected void doAddOnlineSegment(String segmentName)
SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName);
if (segmentDataManager == null) {
addNewOnlineSegment(zkMetadata, indexLoadingConfig);
- return;
- }
- if (segmentDataManager instanceof RealtimeSegmentDataManager) {
+ } else if (segmentDataManager instanceof RealtimeSegmentDataManager) {
_logger.info("Changing segment: {} from CONSUMING to ONLINE", segmentName);
((RealtimeSegmentDataManager) segmentDataManager).goOnlineFromConsuming(zkMetadata);
onConsumingToOnline(segmentName);
- return;
- }
- // For pauseless ingestion, the segment is marked ONLINE before it's built and before the COMMIT_END_METADATA
- // call completes.
- // The server should replace the segment only after the CRC is set by COMMIT_END_METADATA and the segment is
- // marked DONE.
- // This ensures the segment's download URL is available before discarding the locally built copy, preventing
- // data loss if COMMIT_END_METADATA fails.
- if (zkMetadata.getStatus() == Status.DONE) {
+ } else if (zkMetadata.getStatus() == Status.DONE) {
+ // For pauseless ingestion, the segment is marked ONLINE before it's built and before the COMMIT_END_METADATA
+ // call completes.
+ // The server should replace the segment only after the CRC is set by COMMIT_END_METADATA and the segment is
+ // marked DONE.
+ // This ensures the segment's download URL is available before discarding the locally built copy, preventing
+ // data loss if COMMIT_END_METADATA fails.
replaceSegmentIfCrcMismatch(segmentDataManager, zkMetadata, indexLoadingConfig);
}
+ // Register the segment into the consumer coordinator if consumption order is enforced.
+ if (_enforceConsumptionInOrder) {
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
+ if (llcSegmentName != null) {
+ getConsumerCoordinator(llcSegmentName.getPartitionGroupId()).register(llcSegmentName);
Review Comment:
Also this check should be in addSegment() as I don't think when server preloads the segment, this will be called.
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix starvation in consumer lock [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps merged PR #15404:
URL: https://github.com/apache/pinot/pull/15404
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix starvation in consumer lock [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on code in PR #15404:
URL: https://github.com/apache/pinot/pull/15404#discussion_r2019774668
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -474,23 +473,26 @@ protected void doAddOnlineSegment(String segmentName)
SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName);
if (segmentDataManager == null) {
addNewOnlineSegment(zkMetadata, indexLoadingConfig);
- return;
- }
- if (segmentDataManager instanceof RealtimeSegmentDataManager) {
+ } else if (segmentDataManager instanceof RealtimeSegmentDataManager) {
_logger.info("Changing segment: {} from CONSUMING to ONLINE", segmentName);
((RealtimeSegmentDataManager) segmentDataManager).goOnlineFromConsuming(zkMetadata);
onConsumingToOnline(segmentName);
- return;
- }
- // For pauseless ingestion, the segment is marked ONLINE before it's built and before the COMMIT_END_METADATA
- // call completes.
- // The server should replace the segment only after the CRC is set by COMMIT_END_METADATA and the segment is
- // marked DONE.
- // This ensures the segment's download URL is available before discarding the locally built copy, preventing
- // data loss if COMMIT_END_METADATA fails.
- if (zkMetadata.getStatus() == Status.DONE) {
+ } else if (zkMetadata.getStatus() == Status.DONE) {
+ // For pauseless ingestion, the segment is marked ONLINE before it's built and before the COMMIT_END_METADATA
+ // call completes.
+ // The server should replace the segment only after the CRC is set by COMMIT_END_METADATA and the segment is
+ // marked DONE.
+ // This ensures the segment's download URL is available before discarding the locally built copy, preventing
+ // data loss if COMMIT_END_METADATA fails.
replaceSegmentIfCrcMismatch(segmentDataManager, zkMetadata, indexLoadingConfig);
}
+ // Register the segment into the consumer coordinator if consumption order is enforced.
+ if (_enforceConsumptionInOrder) {
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
+ if (llcSegmentName != null) {
+ getConsumerCoordinator(llcSegmentName.getPartitionGroupId()).register(llcSegmentName);
Review Comment:
won't this mess up the watermark if
`consuming -> online` or `offline -> online` transitions are out of order?
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix starvation in consumer lock [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #15404:
URL: https://github.com/apache/pinot/pull/15404#issuecomment-2763022039
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/15404?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `0%` with `78 lines` in your changes missing coverage. Please review.
> Project coverage is 33.79%. Comparing base [(`59551e4`)](https://app.codecov.io/gh/apache/pinot/commit/59551e45224f1535c4863fd577622b37366ccc97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`6014fab`)](https://app.codecov.io/gh/apache/pinot/commit/6014fab3d38caa531abfc62a9ea8553d7b81320b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1946 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/15404?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...ore/data/manager/realtime/ConsumerCoordinator.java](https://app.codecov.io/gh/apache/pinot/pull/15404?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fdata%2Fmanager%2Frealtime%2FConsumerCoordinator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9kYXRhL21hbmFnZXIvcmVhbHRpbWUvQ29uc3VtZXJDb29yZGluYXRvci5qYXZh) | 0.00% | [56 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/15404?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ata/manager/realtime/RealtimeTableDataManager.java](https://app.codecov.io/gh/apache/pinot/pull/15404?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fdata%2Fmanager%2Frealtime%2FRealtimeTableDataManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9kYXRhL21hbmFnZXIvcmVhbHRpbWUvUmVhbHRpbWVUYWJsZURhdGFNYW5hZ2VyLmphdmE=) | 0.00% | [12 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/15404?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...a/manager/realtime/RealtimeSegmentDataManager.java](https://app.codecov.io/gh/apache/pinot/pull/15404?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fdata%2Fmanager%2Frealtime%2FRealtimeSegmentDataManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9kYXRhL21hbmFnZXIvcmVhbHRpbWUvUmVhbHRpbWVTZWdtZW50RGF0YU1hbmFnZXIuamF2YQ==) | 0.00% | [9 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/15404?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../pinot/core/data/manager/BaseTableDataManager.java](https://app.codecov.io/gh/apache/pinot/pull/15404?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fdata%2Fmanager%2FBaseTableDataManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9kYXRhL21hbmFnZXIvQmFzZVRhYmxlRGF0YU1hbmFnZXIuamF2YQ==) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/15404?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (59551e4) and HEAD (6014fab). Click for more details.
>
> <details><summary>HEAD has 48 uploads less than BASE</summary>
>
>| Flag | BASE (59551e4) | HEAD (6014fab) |
>|------|------|------|
>|integration|7|0|
>|integration2|3|0|
>|temurin|12|2|
>|java-21|7|2|
>|skip-bytebuffers-true|3|1|
>|skip-bytebuffers-false|7|1|
>|unittests|5|2|
>|unittests1|2|0|
>|java-11|5|0|
>|unittests2|3|2|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #15404 +/- ##
=============================================
- Coverage 61.75% 33.79% -27.97%
- Complexity 207 684 +477
=============================================
Files 2436 2815 +379
Lines 133233 159560 +26327
Branches 20636 24476 +3840
=============================================
- Hits 82274 53917 -28357
- Misses 44911 101329 +56418
+ Partials 6048 4314 -1734
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.79% <0.00%> (-27.84%)` | :arrow_down: |
| [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.74% <0.00%> (-28.01%)` | :arrow_down: |
| [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.78% <0.00%> (+6.05%)` | :arrow_up: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.79% <0.00%> (-27.97%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.79% <0.00%> (-27.96%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/15404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.79% <0.00%> (+6.05%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/15404?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix starvation in consumer lock [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #15404:
URL: https://github.com/apache/pinot/pull/15404#discussion_r2019672893
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -557,22 +565,10 @@ private void doAddConsumingSegment(String segmentName)
PartitionDedupMetadataManager partitionDedupMetadataManager =
_tableDedupMetadataManager != null ? _tableDedupMetadataManager.getOrCreatePartitionManager(partitionGroupId)
: null;
- RealtimeSegmentDataManager realtimeSegmentDataManager;
- try {
- realtimeSegmentDataManager =
- createRealtimeSegmentDataManager(zkMetadata, tableConfig, indexLoadingConfig, schema, llcSegmentName,
- consumerCoordinator, partitionUpsertMetadataManager, partitionDedupMetadataManager,
- _isTableReadyToConsumeData);
- } catch (SegmentAlreadyConsumedException e) {
Review Comment:
why remove this? without it, we will not unblock OFFLINE -> CONSUMING helix threads for segments that are already in completed state
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix starvation in consumer lock [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix starvation in consumer lock [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #15404:
URL: https://github.com/apache/pinot/pull/15404#issuecomment-2762987862
cc @noob-se7en
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org