xiangfu0 commented on code in PR #19893:
URL: https://github.com/apache/pinot/pull/19893#discussion_r4288267773
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java:
##########
@@ -249,14 +232,26 @@ public void shutDown() {
LOGGER.info("Stopped re-ingestion executor, dropped {} jobs waiting for a
re-ingestion thread", numDroppedJobs);
}
- /// The actual re-ingestion logic, moved into a separate method for clarity.
- /// This is essentially the old synchronous logic you had in reingestSegment.
- private void doReingestSegment(String realtimeTableName, SegmentZKMetadata
segmentZKMetadata,
- IndexLoadingConfig indexLoadingConfig, @Nullable Semaphore
segmentBuildSemaphore)
+ private void doReingestSegment(String realtimeTableName, String segmentName)
throws Exception {
- String segmentName = segmentZKMetadata.getSegmentName();
+ // The table or segment may have been removed or repaired on another
replica while this job was queued.
+ RealtimeTableDataManager tableDataManager =
+ (RealtimeTableDataManager)
_serverInstance.getInstanceDataManager().getTableDataManager(realtimeTableName);
+ if (tableDataManager == null || tableDataManager.isShutDown()) {
+ LOGGER.info("Skipping re-ingestion of segment: {} because table: {} is
no longer available", segmentName,
+ realtimeTableName);
+ return;
+ }
+ SegmentZKMetadata segmentZKMetadata =
tableDataManager.fetchZKMetadataNullable(segmentName);
+ if (segmentZKMetadata == null
+ || segmentZKMetadata.getStatus() != Status.COMMITTING
+ || segmentZKMetadata.getDownloadUrl() != null) {
+ LOGGER.info("Skipping re-ingestion of segment: {} because it was deleted
or no longer needs repair", segmentName);
Review Comment:
MINOR (nit): The single log message covers three different skip reasons
(deleted, not COMMITTING, already has a download URL). For easier debugging of
why a queued job was dropped, you might log which of the three conditions
actually matched.
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java:
##########
@@ -249,14 +232,26 @@ public void shutDown() {
LOGGER.info("Stopped re-ingestion executor, dropped {} jobs waiting for a
re-ingestion thread", numDroppedJobs);
}
- /// The actual re-ingestion logic, moved into a separate method for clarity.
- /// This is essentially the old synchronous logic you had in reingestSegment.
- private void doReingestSegment(String realtimeTableName, SegmentZKMetadata
segmentZKMetadata,
- IndexLoadingConfig indexLoadingConfig, @Nullable Semaphore
segmentBuildSemaphore)
+ private void doReingestSegment(String realtimeTableName, String segmentName)
throws Exception {
- String segmentName = segmentZKMetadata.getSegmentName();
+ // The table or segment may have been removed or repaired on another
replica while this job was queued.
+ RealtimeTableDataManager tableDataManager =
+ (RealtimeTableDataManager)
_serverInstance.getInstanceDataManager().getTableDataManager(realtimeTableName);
+ if (tableDataManager == null || tableDataManager.isShutDown()) {
+ LOGGER.info("Skipping re-ingestion of segment: {} because table: {} is
no longer available", segmentName,
+ realtimeTableName);
+ return;
Review Comment:
MINOR (coverage): This `tableDataManager == null || isShutDown()` skip
branch isn't exercised by the new test. The data provider covers the
stale-metadata cases but not the table-removed / shutdown case, so a regression
here would go unnoticed. Consider adding a case that drops or shuts down the
manager while the job is queued.
--
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]
