Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Pauseless ingestion without failure scenarios #14741

Merged
Prev Previous commit
Fix status check for downloads to also include UPLOADED status
Kartik Khare authored and Kartik Khare committed Jan 10, 2025
commit 81ea9eeb6f26ae91919664bb00eff78b760ae6a2
Original file line number Diff line number Diff line change
@@ -653,7 +653,7 @@ public void reloadSegment(String segmentName, IndexLoadingConfig indexLoadingCon
- Continue loading the segment from the index directory.
*/
boolean shouldDownload =
forceDownload || (zkMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.DONE && !hasSameCRC(
forceDownload || (isSegmentStatusCompleted(zkMetadata) && !hasSameCRC(
zkMetadata, localMetadata));
if (shouldDownload) {
// Create backup directory to handle failure of segment reloading.
@@ -717,6 +717,11 @@ public void reloadSegment(String segmentName, IndexLoadingConfig indexLoadingCon
_logger.info("Reloaded segment: {}", segmentName);
}

private boolean isSegmentStatusCompleted(SegmentZKMetadata zkMetadata) {
return zkMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.DONE
|| zkMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.UPLOADED;
}

private boolean canReuseExistingDirectoryForReload(SegmentZKMetadata segmentZKMetadata, String currentSegmentTier,
SegmentDirectory segmentDirectory, IndexLoadingConfig indexLoadingConfig, Schema schema)
throws Exception {
@@ -1011,8 +1016,7 @@ public boolean tryLoadExistingSegment(SegmentZKMetadata zkMetadata, IndexLoading
Then:
We need to fall back to downloading the segment from deep storage to load it.
*/
if (segmentMetadata == null || (zkMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.DONE
&& !hasSameCRC(zkMetadata, segmentMetadata))) {
if (segmentMetadata == null || (isSegmentStatusCompleted(zkMetadata) && !hasSameCRC(zkMetadata, segmentMetadata))) {
if (segmentMetadata == null) {
_logger.info("Segment: {} does not exist", segmentName);
} else if (!hasSameCRC(zkMetadata, segmentMetadata)) {