Jackie-Jiang commented on code in PR #19893:
URL: https://github.com/apache/pinot/pull/19893#discussion_r4288404757
##########
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:
Added queued-job cases for both a removed table manager and a shut-down
manager in 1b01ccaa32. They verify that the worker skips metadata/config
loading and writer setup, removes the job, and does not increment the failure
metric. All 32 targeted tests pass.
##########
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:
Split the skip guards and added a distinct WARN message for each reason in
1b01ccaa32, including the actual status for non-COMMITTING segments. Skips
still return normally without incrementing the failure metric.
--
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]
