klsince commented on code in PR #13489:
URL: https://github.com/apache/pinot/pull/13489#discussion_r1669423368


##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskExecutor.java:
##########
@@ -60,31 +79,22 @@ protected SegmentConversionResult convert(PinotTaskConfig 
pinotTaskConfig, File
 
     String validDocIdsTypeStr =
         
configs.getOrDefault(MinionConstants.UpsertCompactionTask.VALID_DOC_IDS_TYPE, 
ValidDocIdsType.SNAPSHOT.name());
-    ValidDocIdsType validDocIdsType = 
ValidDocIdsType.valueOf(validDocIdsTypeStr.toUpperCase());
-    ValidDocIdsBitmapResponse validDocIdsBitmapResponse =
-        MinionTaskUtils.getValidDocIdsBitmap(tableNameWithType, segmentName, 
validDocIdsType.toString(),
-            MINION_CONTEXT);
-
-    // Check crc from the downloaded segment against the crc returned from the 
server along with the valid doc id
-    // bitmap. If this doesn't match, this means that we are hitting the race 
condition where the segment has been
-    // uploaded successfully while the server is still reloading the segment. 
Reloading can take a while when the
-    // offheap upsert is used because we will need to delete & add all primary 
keys.
-    // `BaseSingleSegmentConversionExecutor.executeTask()` already checks for 
the crc from the task generator
-    // against the crc from the current segment zk metadata, so we don't need 
to check that here.
     SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
     String originalSegmentCrcFromTaskGenerator = 
configs.get(MinionConstants.ORIGINAL_SEGMENT_CRC_KEY);
     String crcFromDeepStorageSegment = segmentMetadata.getCrc();
-    String crcFromValidDocIdsBitmap = 
validDocIdsBitmapResponse.getSegmentCrc();
-    if (!originalSegmentCrcFromTaskGenerator.equals(crcFromDeepStorageSegment)
-        || 
!originalSegmentCrcFromTaskGenerator.equals(crcFromValidDocIdsBitmap)) {
-      LOGGER.warn("CRC mismatch for segment: {}, expected: {}, actual crc from 
server: {}", segmentName,
-          crcFromDeepStorageSegment, 
validDocIdsBitmapResponse.getSegmentCrc());
-      return new 
SegmentConversionResult.Builder().setTableNameWithType(tableNameWithType).setSegmentName(segmentName)
-          .build();
+    RoaringBitmap validDocIds = 
originalSegmentCrcFromTaskGenerator.equals(crcFromDeepStorageSegment)

Review Comment:
   nit: would suggest to check 
`originalSegmentCrcFromTaskGenerator.equals(crcFromDeepStorageSegment)` firstly 
and log a msg if they are different for a bit more clarify. 



##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtils.java:
##########
@@ -206,4 +183,55 @@ public static boolean 
extractMinionAllowDownloadFromServer(TableConfig tableConf
     }
     return defaultValue;
   }
+
+  /**
+   * Returns the validDocID bitmap from the server whose local segment crc 
matches both crc of ZK metadata and
+   * deepstore copy (expectedCrc).
+   */
+  @Nullable
+  public static RoaringBitmap getValidDocIdFromServerMatchingCrc(String 
tableNameWithType, String segmentName,
+      String validDocIdsType, MinionContext minionContext, String expectedCrc) 
{
+    String clusterName = minionContext.getHelixManager().getClusterName();
+    HelixAdmin helixAdmin = 
minionContext.getHelixManager().getClusterManagmentTool();
+    RoaringBitmap validDocIds = null;
+    List<String> servers = MinionTaskUtils.getServers(segmentName, 
tableNameWithType, helixAdmin, clusterName);
+    for (String server : servers) {
+      InstanceConfig instanceConfig = 
helixAdmin.getInstanceConfig(clusterName, server);
+      String endpoint = InstanceUtils.getServerAdminEndpoint(instanceConfig);
+
+      // We only need aggregated table size and the total number of docs/rows. 
Skipping column related stats, by
+      // passing an empty list.
+      ServerSegmentMetadataReader serverSegmentMetadataReader = new 
ServerSegmentMetadataReader();
+      ValidDocIdsBitmapResponse validDocIdsBitmapResponse;
+      try {
+        validDocIdsBitmapResponse =
+            
serverSegmentMetadataReader.getValidDocIdsBitmapFromServer(tableNameWithType, 
segmentName, endpoint,
+                validDocIdsType, 60_000);
+      } catch (Exception e) {
+        LOGGER.warn(
+            String.format("Unable to retrieve validDocIds bitmap for segment: 
%s from endpoint: %s", segmentName,
+                endpoint), e);
+        continue;
+      }
+
+      // Check crc from the downloaded segment against the crc returned from 
the server along with the valid doc id
+      // bitmap. If this doesn't match, this means that we are hitting the 
race condition where the segment has been
+      // uploaded successfully while the server is still reloading the 
segment. Reloading can take a while when the
+      // offheap upsert is used because we will need to delete & add all 
primary keys.
+      // `BaseSingleSegmentConversionExecutor.executeTask()` already checks 
for the crc from the task generator
+      // against the crc from the current segment zk metadata, so we don't 
need to check that here.
+      String crcFromValidDocIdsBitmap = 
validDocIdsBitmapResponse.getSegmentCrc();
+      if (!expectedCrc.equals(crcFromValidDocIdsBitmap)) {
+        // In this scenario, we are hitting the other replica of the segment 
which did not commit to ZK or deepstore.
+        // We will skip processing this bitmap to query other server to 
confirm if there is a valid matching CRC.
+        String message = String.format("CRC mismatch for segment: %s, expected 
value based on task generator: %s, "
+                + "actual crc from validDocIdsBitmapResponse from endpoint %s: 
%s", segmentName, expectedCrc, endpoint,
+            crcFromValidDocIdsBitmap);
+        LOGGER.warn(message);
+        continue;
+      }
+      validDocIds = 
RoaringBitmapUtils.deserialize(validDocIdsBitmapResponse.getBitmap());

Review Comment:
   `return validDocIds` here to avoid checking following servers?



-- 
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-unsubscr...@pinot.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org

Reply via email to