You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "xiangfu0 (via GitHub)" <gi...@apache.org> on 2025/09/13 16:52:08 UTC
[PR] allow fallback to retain mode if download url is missing for CompletionMode.DOWNLOAD [pinot]
xiangfu0 opened a new pull request, #16810:
URL: https://github.com/apache/pinot/pull/16810
This pull request improves the handling of segment completion and online transitions when using the DOWNLOAD completion mode in the `RealtimeSegmentDataManager`. The main focus is to ensure that the system checks for a valid download URL before attempting to download a segment, and provides clear logging and fallback behavior if the URL is missing or invalid.
**Enhancements to segment completion and transition logic:**
* When in KEEP state and using DOWNLOAD mode, the code now fetches fresh metadata from ZooKeeper to check if a download URL is available. If the URL is missing or invalid, it logs a warning and falls back to retaining the segment instead of attempting a download.
* During the transition from consuming to online, the code checks for the presence of a download URL before downloading a segment. If the URL is missing, it logs a warning and falls back to building the segment locally. [[1]](diffhunk://#diff-fd2e3e273d7742190b27e7b26472460ff5dc6038f895e8647972b205fa8edccaL1463-R1497) [[2]](diffhunk://#diff-fd2e3e273d7742190b27e7b26472460ff5dc6038f895e8647972b205fa8edccaL1513-L1516)
--
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] allow fallback to retain mode if download url is missing for CompletionMode.DOWNLOAD [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #16810:
URL: https://github.com/apache/pinot/pull/16810#discussion_r2346874107
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -850,8 +850,19 @@ public void run() {
break;
case KEEP: {
if (_segmentCompletionMode == CompletionMode.DOWNLOAD) {
- _state = State.DISCARDED;
- break;
+ // Fetch fresh metadata from ZooKeeper to check if download URL has been set by another replica
+ String downloadUrl = _realtimeTableDataManager.fetchZKMetadata(_segmentNameStr).getDownloadUrl();
+ if (StringUtils.isNotEmpty(downloadUrl)) {
+ _segmentLogger.info(
+ "CompletionMode is DOWNLOAD and download URL is available for segment: {}. URL: {}",
+ _segmentNameStr, downloadUrl);
+ _state = State.DISCARDED;
+ break;
+ } else {
+ _segmentLogger.warn(
+ "CompletionMode is DOWNLOAD but download URL is missing segment: {}. "
Review Comment:
The error message is missing the word "for" before "segment". It should read "missing for segment: {}" to be grammatically correct.
```suggestion
"CompletionMode is DOWNLOAD but download URL is missing for segment: {}. "
```
--
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] allow fallback to retain mode if download url is missing for CompletionMode.DOWNLOAD [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #16810:
URL: https://github.com/apache/pinot/pull/16810#discussion_r2346868358
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1460,58 +1471,59 @@ public void goOnlineFromConsuming(SegmentZKMetadata segmentZKMetadata)
case CATCHING_UP:
case HOLDING:
case INITIAL_CONSUMING:
- switch (_segmentCompletionMode) {
- case DOWNLOAD:
- _segmentLogger.info("State {}. CompletionMode {}. Downloading to replace", _state.toString(),
- _segmentCompletionMode);
+ if (_segmentCompletionMode == CompletionMode.DOWNLOAD) {
+ // Fetch fresh metadata from ZooKeeper to check if download URL has been set by another replica
+ String downloadUrl = segmentZKMetadata.getDownloadUrl();
+ if (StringUtils.isNotEmpty(downloadUrl)) {
+ _segmentLogger.info("State {}. CompletionMode {}. Downloading to replace with fresh metadata. URL: {}",
+ _state.toString(), _segmentCompletionMode, downloadUrl);
downloadSegmentAndReplace(segmentZKMetadata);
break;
- case DEFAULT:
- // Allow to catch up upto final offset, and then replace.
- if (_currentOffset.compareTo(endOffset) > 0) {
- // We moved ahead of the offset that is committed in ZK.
- _segmentLogger
- .warn("Current offset {} ahead of the offset in zk {}. Downloading to replace", _currentOffset,
- endOffset);
- downloadSegmentAndReplace(segmentZKMetadata);
- } else if (_currentOffset.compareTo(endOffset) == 0) {
- _segmentLogger
- .info("Current offset {} matches offset in zk {}. Replacing segment", _currentOffset, endOffset);
- if (!buildSegmentAndReplace()) {
- _segmentLogger.warn("Failed to build the segment: {} and replace. Downloading to replace",
- _segmentNameStr);
- downloadSegmentAndReplace(segmentZKMetadata);
- }
- } else {
- boolean success = false;
- // Since online helix transition for a segment can arrive before segment's consumer acquires the
- // semaphore, check _consumerSemaphoreAcquired before catching up.
- // This is to avoid consuming in parallel to another segment's consumer.
- if (_consumerSemaphoreAcquired.get()) {
- _segmentLogger.info("Attempting to catch up from offset {} to {} ", _currentOffset, endOffset);
- success = catchupToFinalOffset(endOffset,
- TimeUnit.MILLISECONDS.convert(MAX_TIME_FOR_CONSUMING_TO_ONLINE_IN_SECONDS, TimeUnit.SECONDS));
- } else {
- _segmentLogger.warn("Consumer semaphore was not acquired, Skipping catch up from offset {} to {} ",
- _currentOffset, endOffset);
- }
+ } else {
+ _segmentLogger.warn("State {}. CompletionMode is DOWNLOAD but download URL is missing. "
+ + "Falling back to building segment locally for segment: {}", _state.toString(), _segmentNameStr);
+ }
+ }
Review Comment:
The comment mentions fetching fresh metadata from ZooKeeper, but the code uses the passed `segmentZKMetadata` parameter instead of fetching fresh metadata like in the KEEP case above. This inconsistency could lead to using stale metadata when the download URL might have been updated by another replica.
##########
pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java:
##########
@@ -859,6 +859,159 @@ public void testShouldNotSkipUnfilteredMessagesIfNotIndexedAndRowCountThresholdI
}
}
+ @Test
+ public void testCompletionModeDownloadWithUrlValidation()
+ throws Exception {
+ // Test the new validation logic for CompletionMode.DOWNLOAD
+
+ // Test Case 1: Valid download URL - should proceed with DOWNLOAD mode
+ testCompletionModeDownloadWithValidUrl();
+
+ // Test Case 2: Missing/null download URL - should fall back to RETAINING mode
+ testCompletionModeDownloadWithMissingUrl();
+
+ // Test Case 3: Empty download URL - should fall back to RETAINING mode
+ testCompletionModeDownloadWithEmptyUrl();
+ }
+
+ private void testCompletionModeDownloadWithValidUrl()
+ throws Exception {
+ String downloadUrl = "http://example.com/segment.tar.gz";
+ TestResult expectedResult = new TestResult(
+ RealtimeSegmentDataManager.State.DISCARDED, false);
+
+ runCompletionModeDownloadTest(downloadUrl, downloadUrl, expectedResult);
+ }
+
+ private void testCompletionModeDownloadWithMissingUrl()
+ throws Exception {
+ TestResult expectedResult = new TestResult(
+ RealtimeSegmentDataManager.State.RETAINED, true);
+
+ runCompletionModeDownloadTest(null, null, expectedResult);
+ }
+
+ private void testCompletionModeDownloadWithEmptyUrl()
+ throws Exception {
+ TestResult expectedResult = new TestResult(
+ RealtimeSegmentDataManager.State.RETAINED, true);
+
+ runCompletionModeDownloadTest("", "", expectedResult);
+ }
+
+ /**
+ * Helper method to run completion mode download tests with different download URL scenarios.
+ *
+ * @param freshMetadataDownloadUrl The download URL returned by fetchZKMetadata (fresh metadata)
+ * @param initialSegmentDownloadUrl The download URL in the initial segment metadata
+ * @param expectedResult The expected test result (state and buildAndReplaceCalled flag)
+ */
+ private void runCompletionModeDownloadTest(String freshMetadataDownloadUrl,
+ String initialSegmentDownloadUrl, TestResult expectedResult)
+ throws Exception {
+
+ // Create table config with DOWNLOAD completion mode
+ TableConfig tableConfig = createTableConfigWithDownloadCompletionMode();
+
+ // Create mock table data manager with the specified download URL
+ RealtimeTableDataManager mockTableDataManager = createMockTableDataManager(freshMetadataDownloadUrl);
+
+ // Create segment data manager with DOWNLOAD completion mode
+ SegmentZKMetadata segmentZKMetadata = createZkMetadata();
+ if (initialSegmentDownloadUrl != null) {
+ segmentZKMetadata.setDownloadUrl(initialSegmentDownloadUrl);
+ }
+
+ LLCSegmentName llcSegmentName = new LLCSegmentName(SEGMENT_NAME_STR);
+ Schema schema = Fixtures.createSchema();
+ ServerMetrics serverMetrics = new ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry());
+
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ new FakeRealtimeSegmentDataManager(segmentZKMetadata, tableConfig, mockTableDataManager,
+ new File(TEMP_DIR, REALTIME_TABLE_NAME).getAbsolutePath(), schema, llcSegmentName,
+ _partitionGroupIdToConsumerCoordinatorMap, serverMetrics, new TimeSupplier())) {
+
+ // Execute the consumer with KEEP response
+ executeConsumerWithKeepResponse(segmentDataManager);
+
+ // Verify the expected results
+ Assert.assertEquals(segmentDataManager._state.get(segmentDataManager),
+ expectedResult.expectedState);
+ Assert.assertEquals(segmentDataManager._buildAndReplaceCalled,
+ expectedResult.expectedBuildAndReplaceCalled);
+ Assert.assertTrue(segmentDataManager._responses.isEmpty());
+ Assert.assertTrue(segmentDataManager._consumeOffsets.isEmpty());
+ }
+ }
+
+ /**
+ * Creates a table config with DOWNLOAD completion mode.
+ */
+ private TableConfig createTableConfigWithDownloadCompletionMode()
+ throws Exception {
+ TableConfig tableConfig = createTableConfig();
+ tableConfig.getValidationConfig().setCompletionConfig(
+ new org.apache.pinot.spi.config.table.CompletionConfig("DOWNLOAD"));
+ return tableConfig;
+ }
+
+ /**
+ * Creates a mock table data manager that returns the specified download URL in fresh metadata.
+ */
+ private RealtimeTableDataManager createMockTableDataManager(String downloadUrl) {
+ RealtimeTableDataManager mockTableDataManager = mock(RealtimeTableDataManager.class);
+ when(mockTableDataManager.getInstanceId()).thenReturn("server-1");
+ when(mockTableDataManager.getSegmentLock(any())).thenReturn(mock(Lock.class));
+
+ // Mock fetchZKMetadata to return metadata with the specified download URL
+ SegmentZKMetadata metadata = createZkMetadata();
+ metadata.setDownloadUrl(downloadUrl);
+ when(mockTableDataManager.fetchZKMetadata(SEGMENT_NAME_STR)).thenReturn(metadata);
+
+ // Set up stats history and consumer directory
+ RealtimeSegmentStatsHistory statsHistory = mock(RealtimeSegmentStatsHistory.class);
+ when(statsHistory.getEstimatedCardinality(anyString())).thenReturn(200);
+ when(statsHistory.getEstimatedAvgColSize(anyString())).thenReturn(32);
+ when(mockTableDataManager.getStatsHistory()).thenReturn(statsHistory);
+ when(mockTableDataManager.getConsumerDir()).thenReturn(TEMP_DIR.getAbsolutePath() + "/consumerDir");
+
+ // Set up consumer coordinator in the map
+ _partitionGroupIdToConsumerCoordinatorMap.putIfAbsent(PARTITION_GROUP_ID,
+ new ConsumerCoordinator(false, mockTableDataManager));
+
+ return mockTableDataManager;
+ }
+
+ /**
+ * Executes the consumer with a KEEP response to trigger the completion mode logic.
+ */
+ private void executeConsumerWithKeepResponse(FakeRealtimeSegmentDataManager segmentDataManager) {
+ RealtimeSegmentDataManager.PartitionConsumer consumer = segmentDataManager.createPartitionConsumer();
+ final LongMsgOffset endOffset = new LongMsgOffset(START_OFFSET_VALUE + 500);
+ segmentDataManager._consumeOffsets.add(endOffset);
+
+ final SegmentCompletionProtocol.Response response = new SegmentCompletionProtocol.Response(
+ new SegmentCompletionProtocol.Response.Params().withStatus(
+ SegmentCompletionProtocol.ControllerResponseStatus.KEEP)
+ .withStreamPartitionMsgOffset(endOffset.toString()));
+ segmentDataManager._responses.add(response);
+
+ consumer.run();
+ }
+
+ /**
+ * Helper class to encapsulate expected test results.
+ */
+ private static class TestResult {
+ final RealtimeSegmentDataManager.State expectedState;
+ final boolean expectedBuildAndReplaceCalled;
+
Review Comment:
The TestResult class lacks Javadoc documentation. Since this is a helper class with specific fields that represent test expectations, it should include documentation explaining the purpose of each field and the class itself.
```suggestion
/**
* Helper class to encapsulate expected test results for segment state transitions.
* <p>
* This class holds the expected state of a {@link RealtimeSegmentDataManager} after a test,
* as well as whether the build-and-replace operation was expected to be called.
*/
private static class TestResult {
/**
* The expected state of the {@link RealtimeSegmentDataManager} after the test execution.
*/
final RealtimeSegmentDataManager.State expectedState;
/**
* Whether the build-and-replace operation was expected to be called during the test.
*/
final boolean expectedBuildAndReplaceCalled;
/**
* Constructs a TestResult with the expected state and build-and-replace flag.
*
* @param expectedState The expected state of the segment manager after the test.
* @param expectedBuildAndReplaceCalled Whether build-and-replace was expected to 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] allow fallback to retain mode if download url is missing for CompletionMode.DOWNLOAD [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #16810:
URL: https://github.com/apache/pinot/pull/16810#discussion_r2346836840
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -850,8 +850,19 @@ public void run() {
break;
case KEEP: {
if (_segmentCompletionMode == CompletionMode.DOWNLOAD) {
- _state = State.DISCARDED;
- break;
+ // Fetch fresh metadata from ZooKeeper to check if download URL has been set by another replica
+ String downloadUrl = _realtimeTableDataManager.fetchZKMetadata(_segmentNameStr).getDownloadUrl();
+ if (StringUtils.isNotEmpty(downloadUrl)) {
+ _segmentLogger.info(
+ "CompletionMode is DOWNLOAD and download URL is available for segment: {}. URL: {}",
+ _segmentNameStr, downloadUrl);
+ _state = State.DISCARDED;
+ break;
+ } else {
+ _segmentLogger.warn(
+ "CompletionMode is DOWNLOAD but download URL is missing or invalid for segment: {}. "
Review Comment:
[nitpick] The log message mentions 'invalid' download URL but the code only checks if it's empty/null with StringUtils.isNotEmpty(). Consider clarifying the log message to only mention 'missing' or add validation for invalid URLs.
```suggestion
"CompletionMode is DOWNLOAD but download URL is missing for segment: {}. "
```
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -850,8 +850,19 @@ public void run() {
break;
case KEEP: {
if (_segmentCompletionMode == CompletionMode.DOWNLOAD) {
- _state = State.DISCARDED;
- break;
+ // Fetch fresh metadata from ZooKeeper to check if download URL has been set by another replica
+ String downloadUrl = _realtimeTableDataManager.fetchZKMetadata(_segmentNameStr).getDownloadUrl();
Review Comment:
Fetching metadata from ZooKeeper on every KEEP operation could introduce performance overhead. Consider caching the metadata or checking if the segment metadata has been updated since the last check to avoid unnecessary ZK calls.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1460,58 +1472,60 @@ public void goOnlineFromConsuming(SegmentZKMetadata segmentZKMetadata)
case CATCHING_UP:
case HOLDING:
case INITIAL_CONSUMING:
- switch (_segmentCompletionMode) {
- case DOWNLOAD:
- _segmentLogger.info("State {}. CompletionMode {}. Downloading to replace", _state.toString(),
- _segmentCompletionMode);
+ if (_segmentCompletionMode == CompletionMode.DOWNLOAD) {
+ // Fetch fresh metadata from ZooKeeper to check if download URL has been set by another replica
+ String downloadUrl = segmentZKMetadata.getDownloadUrl();
+ if (StringUtils.isNotEmpty(downloadUrl)) {
+ _segmentLogger.info("State {}. CompletionMode {}. Downloading to replace with fresh metadata. URL: {}",
+ _state.toString(), _segmentCompletionMode, downloadUrl);
downloadSegmentAndReplace(segmentZKMetadata);
break;
- case DEFAULT:
- // Allow to catch up upto final offset, and then replace.
- if (_currentOffset.compareTo(endOffset) > 0) {
- // We moved ahead of the offset that is committed in ZK.
- _segmentLogger
- .warn("Current offset {} ahead of the offset in zk {}. Downloading to replace", _currentOffset,
- endOffset);
- downloadSegmentAndReplace(segmentZKMetadata);
- } else if (_currentOffset.compareTo(endOffset) == 0) {
- _segmentLogger
- .info("Current offset {} matches offset in zk {}. Replacing segment", _currentOffset, endOffset);
- if (!buildSegmentAndReplace()) {
- _segmentLogger.warn("Failed to build the segment: {} and replace. Downloading to replace",
- _segmentNameStr);
- downloadSegmentAndReplace(segmentZKMetadata);
- }
- } else {
- boolean success = false;
- // Since online helix transition for a segment can arrive before segment's consumer acquires the
- // semaphore, check _consumerSemaphoreAcquired before catching up.
- // This is to avoid consuming in parallel to another segment's consumer.
- if (_consumerSemaphoreAcquired.get()) {
- _segmentLogger.info("Attempting to catch up from offset {} to {} ", _currentOffset, endOffset);
- success = catchupToFinalOffset(endOffset,
- TimeUnit.MILLISECONDS.convert(MAX_TIME_FOR_CONSUMING_TO_ONLINE_IN_SECONDS, TimeUnit.SECONDS));
- } else {
- _segmentLogger.warn("Consumer semaphore was not acquired, Skipping catch up from offset {} to {} ",
- _currentOffset, endOffset);
- }
+ } else {
+ _segmentLogger.warn("State {}. CompletionMode is DOWNLOAD but download URL is missing: {}. "
+ + "Falling back to building segment locally. Download URL: {}", _state.toString(),
+ _segmentNameStr, downloadUrl);
Review Comment:
[nitpick] The log message format includes a colon after 'missing' but then uses a period, creating inconsistent punctuation. Also, the placeholder order doesn't match the arguments - the log suggests the URL comes after the segment name, but the arguments are state, segment name, then URL.
```suggestion
_segmentLogger.warn("State {}. CompletionMode is DOWNLOAD but download URL is missing ({}). "
+ "Falling back to building segment locally for segment: {}", _state.toString(),
downloadUrl, _segmentNameStr);
```
--
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] allow fallback to retain mode if download url is missing for CompletionMode.DOWNLOAD [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #16810:
URL: https://github.com/apache/pinot/pull/16810#discussion_r2370266998
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -850,8 +850,19 @@ public void run() {
break;
case KEEP: {
if (_segmentCompletionMode == CompletionMode.DOWNLOAD) {
- _state = State.DISCARDED;
- break;
+ // Fetch fresh metadata from ZooKeeper to check if download URL has been set by another replica
+ String downloadUrl = _realtimeTableDataManager.fetchZKMetadata(_segmentNameStr).getDownloadUrl();
Review Comment:
What about peer download?
--
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] allow fallback to retain mode if download url is missing for CompletionMode.DOWNLOAD [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #16810:
URL: https://github.com/apache/pinot/pull/16810#issuecomment-3288729281
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/16810?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `71.42857%` with `10 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 56.29%. Comparing base ([`19ae6b9`](https://app.codecov.io/gh/apache/pinot/commit/19ae6b9e64a59d5e15d93454387e45899b4a982e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`d3a3172`](https://app.codecov.io/gh/apache/pinot/commit/d3a31721d8cef553c0378c08296d035e2bbebe5f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/16810?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...a/manager/realtime/RealtimeSegmentDataManager.java](https://app.codecov.io/gh/apache/pinot/pull/16810?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==) | 71.42% | [8 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16810?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 (19ae6b9) and HEAD (d3a3172). Click for more details.
>
> <details><summary>HEAD has 32 uploads less than BASE</summary>
>
>| Flag | BASE (19ae6b9) | HEAD (d3a3172) |
>|------|------|------|
>|java-21|4|1|
>|unittests1|2|1|
>|unittests|4|1|
>|temurin|9|1|
>|java-11|5|0|
>|unittests2|2|0|
>|integration|5|0|
>|integration1|2|0|
>|custom-integration1|2|0|
>|integration2|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #16810 +/- ##
============================================
- Coverage 63.36% 56.29% -7.07%
+ Complexity 1400 700 -700
============================================
Files 3055 2396 -659
Lines 178934 133184 -45750
Branches 27423 21191 -6232
============================================
- Hits 113381 74980 -38401
+ Misses 56805 51955 -4850
+ Partials 8748 6249 -2499
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/16810/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/16810/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/16810/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/16810/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/16810/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/16810/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/16810/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.29% <71.42%> (-7.04%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/16810/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.29% <71.42%> (-7.07%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/16810/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.29% <71.42%> (-7.07%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/16810/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.29% <71.42%> (-0.01%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/16810/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
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/16810?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>
- :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] allow fallback to retain mode if download url is missing for CompletionMode.DOWNLOAD [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #16810:
URL: https://github.com/apache/pinot/pull/16810#discussion_r2430160467
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1650,4 +1650,11 @@ private static void closeSegmentDirectoryQuietly(SegmentDirectory segmentDirecto
}
}
}
+
+ /**
+ * Returns the configured peer download scheme if peer-to-peer download is enabled; otherwise null.
+ */
+ public String getPeerDownloadScheme() {
Review Comment:
Annotate it with `@Nullable`
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -583,6 +583,21 @@ public File downloadSegment(SegmentZKMetadata zkMetadata)
if (status.isCompleted()) {
// Segment is completed and ready to be downloaded either from deep storage or from a peer (if peer-to-peer
// download is enabled).
+ String downloadUrl = zkMetadata.getDownloadUrl();
+ if (StringUtils.isNotEmpty(downloadUrl)) {
+ return super.downloadSegment(zkMetadata);
+ }
+ // Download URL might be missing due to race conditions or controller commit failure. If peer download
+ // is enabled, try to fetch from ONLINE peers instead of failing fast.
+ if (_peerDownloadScheme != null) {
+ try {
+ return downloadSegmentFromPeers(zkMetadata);
+ } catch (Exception e) {
+ _logger.warn("Peer download attempt failed for completed segment: {}. Falling back to deep store if possible",
+ zkMetadata.getSegmentName(), e);
+ }
+ }
+ // Fall back to the default behavior which will validate and potentially throw if URL is still missing.
Review Comment:
This part shouldn't be required. The base class can handle completed segment download
--
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] allow fallback to retain mode if download url is missing for CompletionMode.DOWNLOAD [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 merged PR #16810:
URL: https://github.com/apache/pinot/pull/16810
--
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] allow fallback to retain mode if download url is missing for CompletionMode.DOWNLOAD [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #16810:
URL: https://github.com/apache/pinot/pull/16810#discussion_r2346865132
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1379,6 +1390,7 @@ private static class ConsumptionStopIndicator {
final Logger _logger;
final ServerSegmentCompletionProtocolHandler _protocolHandler;
final String _reason;
+
Review Comment:
This empty line addition appears unintentional and doesn't follow the existing code formatting patterns. Remove the extra blank line to maintain consistency with the surrounding code.
##########
pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java:
##########
@@ -859,6 +859,202 @@ public void testShouldNotSkipUnfilteredMessagesIfNotIndexedAndRowCountThresholdI
}
}
+ @Test
+ public void testCompletionModeDownloadWithUrlValidation()
+ throws Exception {
+ // Test the new validation logic for CompletionMode.DOWNLOAD
+
+ // Test Case 1: Valid download URL - should proceed with DOWNLOAD mode
+ testCompletionModeDownloadWithValidUrl();
+
+ // Test Case 2: Missing/null download URL - should fall back to RETAINING mode
+ testCompletionModeDownloadWithMissingUrl();
+
+ // Test Case 3: Empty download URL - should fall back to RETAINING mode
+ testCompletionModeDownloadWithEmptyUrl();
+ }
+
+ private void testCompletionModeDownloadWithValidUrl()
+ throws Exception {
+ // Create table config with DOWNLOAD completion mode
+ TableConfig tableConfig = createTableConfig();
+ tableConfig.getValidationConfig().setCompletionConfig(
+ new org.apache.pinot.spi.config.table.CompletionConfig("DOWNLOAD"));
+
+ // Create mock table data manager that returns valid download URL
+ RealtimeTableDataManager mockTableDataManager = mock(RealtimeTableDataManager.class);
+ when(mockTableDataManager.getInstanceId()).thenReturn("server-1");
+ when(mockTableDataManager.getSegmentLock(any())).thenReturn(mock(Lock.class));
+
+ // Mock fetchZKMetadata to return metadata with valid download URL
+ SegmentZKMetadata metadataWithUrl = createZkMetadata();
+ metadataWithUrl.setDownloadUrl("http://example.com/segment.tar.gz");
+ when(mockTableDataManager.fetchZKMetadata(SEGMENT_NAME_STR)).thenReturn(metadataWithUrl);
+
+ RealtimeSegmentStatsHistory statsHistory = mock(RealtimeSegmentStatsHistory.class);
+ when(statsHistory.getEstimatedCardinality(anyString())).thenReturn(200);
+ when(statsHistory.getEstimatedAvgColSize(anyString())).thenReturn(32);
+ when(mockTableDataManager.getStatsHistory()).thenReturn(statsHistory);
+ when(mockTableDataManager.getConsumerDir()).thenReturn(TEMP_DIR.getAbsolutePath() + "/consumerDir");
+
+ // Set up consumer coordinator in the map
+ _partitionGroupIdToConsumerCoordinatorMap.putIfAbsent(PARTITION_GROUP_ID,
+ new ConsumerCoordinator(false, mockTableDataManager));
+
+ // Create segment data manager with DOWNLOAD completion mode
+ SegmentZKMetadata segmentZKMetadata = createZkMetadata();
+ segmentZKMetadata.setDownloadUrl("http://example.com/segment.tar.gz"); // Initial URL
+ LLCSegmentName llcSegmentName = new LLCSegmentName(SEGMENT_NAME_STR);
+ Schema schema = Fixtures.createSchema();
+ ServerMetrics serverMetrics = new ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry());
+
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ new FakeRealtimeSegmentDataManager(segmentZKMetadata, tableConfig, mockTableDataManager,
+ new File(TEMP_DIR, REALTIME_TABLE_NAME).getAbsolutePath(), schema, llcSegmentName,
+ _partitionGroupIdToConsumerCoordinatorMap, serverMetrics, new TimeSupplier())) {
+
+ // Set up the consumer to get KEEP response
+ RealtimeSegmentDataManager.PartitionConsumer consumer = segmentDataManager.createPartitionConsumer();
+ final LongMsgOffset endOffset = new LongMsgOffset(START_OFFSET_VALUE + 500);
+ segmentDataManager._consumeOffsets.add(endOffset);
+
+ final SegmentCompletionProtocol.Response response = new SegmentCompletionProtocol.Response(
+ new SegmentCompletionProtocol.Response.Params().withStatus(
+ SegmentCompletionProtocol.ControllerResponseStatus.KEEP)
+ .withStreamPartitionMsgOffset(endOffset.toString()));
+ segmentDataManager._responses.add(response);
+
+ consumer.run();
+
+ // Verify that with valid download URL, we proceed to DISCARDED state (DOWNLOAD mode)
+ Assert.assertEquals(segmentDataManager._state.get(segmentDataManager),
+ RealtimeSegmentDataManager.State.DISCARDED);
+ Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
+ Assert.assertTrue(segmentDataManager._responses.isEmpty());
+ Assert.assertTrue(segmentDataManager._consumeOffsets.isEmpty());
+ }
+ }
+
+ private void testCompletionModeDownloadWithMissingUrl()
+ throws Exception {
+ // Create table config with DOWNLOAD completion mode
+ TableConfig tableConfig = createTableConfig();
+ tableConfig.getValidationConfig().setCompletionConfig(
+ new org.apache.pinot.spi.config.table.CompletionConfig("DOWNLOAD"));
+
+ // Create mock table data manager that returns metadata with null download URL
+ RealtimeTableDataManager mockTableDataManager = mock(RealtimeTableDataManager.class);
+ when(mockTableDataManager.getInstanceId()).thenReturn("server-1");
+ when(mockTableDataManager.getSegmentLock(any())).thenReturn(mock(Lock.class));
+
+ // Mock fetchZKMetadata to return metadata with null download URL
+ SegmentZKMetadata metadataWithoutUrl = createZkMetadata();
+ metadataWithoutUrl.setDownloadUrl(null); // No download URL
+ when(mockTableDataManager.fetchZKMetadata(SEGMENT_NAME_STR)).thenReturn(metadataWithoutUrl);
Review Comment:
There's significant code duplication across the three test methods (testCompletionModeDownloadWithValidUrl, testCompletionModeDownloadWithMissingUrl, testCompletionModeDownloadWithEmptyUrl). Consider extracting common setup logic into a helper method to reduce duplication and improve maintainability.
--
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] allow fallback to retain mode if download url is missing for CompletionMode.DOWNLOAD [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #16810:
URL: https://github.com/apache/pinot/pull/16810#discussion_r2346879266
##########
pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java:
##########
@@ -859,6 +859,174 @@ public void testShouldNotSkipUnfilteredMessagesIfNotIndexedAndRowCountThresholdI
}
}
+ @Test
+ public void testCompletionModeDownloadWithUrlValidation()
+ throws Exception {
+ // Test the new validation logic for CompletionMode.DOWNLOAD
+
+ // Test Case 1: Valid download URL - should proceed with DOWNLOAD mode
+ testCompletionModeDownloadWithValidUrl();
+
+ // Test Case 2: Missing/null download URL - should fall back to RETAINING mode
+ testCompletionModeDownloadWithMissingUrl();
+
+ // Test Case 3: Empty download URL - should fall back to RETAINING mode
+ testCompletionModeDownloadWithEmptyUrl();
+ }
+
+ private void testCompletionModeDownloadWithValidUrl()
+ throws Exception {
+ String downloadUrl = "http://example.com/segment.tar.gz";
Review Comment:
[nitpick] Using 'example.com' in tests is good practice, but consider using a constant or a more descriptive URL that clearly indicates this is a test URL to avoid any confusion.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1460,58 +1471,59 @@ public void goOnlineFromConsuming(SegmentZKMetadata segmentZKMetadata)
case CATCHING_UP:
case HOLDING:
case INITIAL_CONSUMING:
- switch (_segmentCompletionMode) {
- case DOWNLOAD:
- _segmentLogger.info("State {}. CompletionMode {}. Downloading to replace", _state.toString(),
- _segmentCompletionMode);
+ if (_segmentCompletionMode == CompletionMode.DOWNLOAD) {
+ // Check if download URL has been set by another replica
+ String downloadUrl = segmentZKMetadata.getDownloadUrl();
+ if (StringUtils.isNotEmpty(downloadUrl)) {
+ _segmentLogger.info("State {}. CompletionMode {}. Downloading to replace with fresh metadata. URL: {}",
+ _state.toString(), _segmentCompletionMode, downloadUrl);
downloadSegmentAndReplace(segmentZKMetadata);
break;
- case DEFAULT:
- // Allow to catch up upto final offset, and then replace.
- if (_currentOffset.compareTo(endOffset) > 0) {
- // We moved ahead of the offset that is committed in ZK.
- _segmentLogger
- .warn("Current offset {} ahead of the offset in zk {}. Downloading to replace", _currentOffset,
- endOffset);
- downloadSegmentAndReplace(segmentZKMetadata);
- } else if (_currentOffset.compareTo(endOffset) == 0) {
- _segmentLogger
- .info("Current offset {} matches offset in zk {}. Replacing segment", _currentOffset, endOffset);
- if (!buildSegmentAndReplace()) {
- _segmentLogger.warn("Failed to build the segment: {} and replace. Downloading to replace",
- _segmentNameStr);
- downloadSegmentAndReplace(segmentZKMetadata);
- }
- } else {
- boolean success = false;
- // Since online helix transition for a segment can arrive before segment's consumer acquires the
- // semaphore, check _consumerSemaphoreAcquired before catching up.
- // This is to avoid consuming in parallel to another segment's consumer.
- if (_consumerSemaphoreAcquired.get()) {
- _segmentLogger.info("Attempting to catch up from offset {} to {} ", _currentOffset, endOffset);
- success = catchupToFinalOffset(endOffset,
- TimeUnit.MILLISECONDS.convert(MAX_TIME_FOR_CONSUMING_TO_ONLINE_IN_SECONDS, TimeUnit.SECONDS));
- } else {
- _segmentLogger.warn("Consumer semaphore was not acquired, Skipping catch up from offset {} to {} ",
- _currentOffset, endOffset);
- }
+ } else {
+ _segmentLogger.warn("State {}. CompletionMode is DOWNLOAD but download URL is missing. "
+ + "Falling back to building segment locally for segment: {}", _state.toString(), _segmentNameStr);
+ }
+ }
+ // Allow to catch up upto final offset, and then replace.
+ if (_currentOffset.compareTo(endOffset) > 0) {
+ // We moved ahead of the offset that is committed in ZK.
+ _segmentLogger.warn("Current offset {} ahead of the offset in zk {}. Downloading to replace",
+ _currentOffset, endOffset);
Review Comment:
[nitpick] The code formatting change from chained method calls to line breaks seems unnecessary and inconsistent with other similar log statements in the codebase. Consider keeping the original formatting to maintain consistency.
```suggestion
_segmentLogger.warn("Current offset {} ahead of the offset in zk {}. Downloading to replace", _currentOffset, endOffset);
```
--
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