Fix crc mismatch during deepstore upload retry task

pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

@@ -82,6 +82,7 @@
 import org.apache.pinot.controller.helix.core.realtime.segment.FlushThresholdUpdater;
 import org.apache.pinot.controller.helix.core.retention.strategy.RetentionStrategy;
 import org.apache.pinot.controller.helix.core.retention.strategy.TimeRetentionStrategy;
+import org.apache.pinot.controller.util.ServerSegmentMetadataReader;
 import org.apache.pinot.controller.validation.RealtimeSegmentValidationManager;
 import org.apache.pinot.core.data.manager.realtime.SegmentCompletionUtils;
 import org.apache.pinot.core.util.PeerServerSegmentFinder;
@@ -1555,9 +1556,20 @@ public void uploadToDeepStoreIfMissing(TableConfig tableConfig, List<SegmentZKMe
                     segmentName));
           }
 
-          // Randomly ask one server to upload
-          URI uri = peerSegmentURIs.get(RANDOM.nextInt(peerSegmentURIs.size()));
-          String serverUploadRequestUrl = StringUtil.join("/", uri.toString(), "upload");
+          ServerSegmentMetadataReader serverSegmentMetadataReader = new ServerSegmentMetadataReader();
+          Collections.shuffle(peerSegmentURIs);
+          URI uriToUpload = peerSegmentURIs.get(0);
+          for (URI uri : peerSegmentURIs) {
+            uriToUpload = uri;
+            LOGGER.info("Get CRC from server {} for LLC segment {}", uri, segmentName);
+            String crcFromServer = serverSegmentMetadataReader.getCrcForSegmentFromServer(realtimeTableName,
+                segmentName, uri.toString());
+            if (crcFromServer != null && Long.parseLong(crcFromServer) == segmentZKMetadata.getCrc()) {
+              break;
+            }
+          }
+
+          String serverUploadRequestUrl = StringUtil.join("/", uriToUpload.toString(), "upload");
           serverUploadRequestUrl =
               String.format("%s?uploadTimeoutMs=%d", serverUploadRequestUrl, _deepstoreUploadRetryTimeoutMs);
           LOGGER.info("Ask server to upload LLC segment {} to deep store by this path: {}", segmentName,

pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java

@@ -397,6 +397,24 @@ public ValidDocIdsBitmapResponse getValidDocIdsBitmapFromServer(String tableName
     return response;
   }
 
+  /**
+   * Returns the crc value of the segment hosted on the server.
+   */
+  @Nullable
+  public String getCrcForSegmentFromServer(String tableNameWithType, String segmentName, String endpoint) {
+    try {
+      // build the url
+      tableNameWithType = URLEncoder.encode(tableNameWithType, StandardCharsets.UTF_8);
+      segmentName = URLEncoder.encode(segmentName, StandardCharsets.UTF_8);
+      String url = String.format("%s/segments/%s/%s/crc", endpoint, tableNameWithType, segmentName);
+      ClientConfig clientConfig = new ClientConfig();
+      return ClientBuilder.newClient(clientConfig).target(url).request(MediaType.APPLICATION_JSON).get(String.class);
+    } catch (Exception e) {
+      LOGGER.error("Error in fetching crc from server {} for segment {}: {}", endpoint, segmentName, e.getMessage());
+    }
+    return null;
+  }
+
   private String generateAggregateSegmentMetadataServerURL(String tableNameWithType, List<String> columns,
       String endpoint) {
     tableNameWithType = URLEncoder.encode(tableNameWithType, StandardCharsets.UTF_8);

pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java

@@ -542,6 +542,35 @@ public ValidDocIdsBitmapResponse downloadValidDocIdsBitmap(
     }
   }
 
+  @GET
+  @Produces(MediaType.APPLICATION_JSON)
+  @Path("/segments/{tableNameWithType}/{segmentName}/crc")
+  @ApiOperation(value = "Return crc value for segment in the server.", notes =
+      "Return crc value for segment in the server.")
+  public String getCrcForSegment(
+      @ApiParam(value = "Name of the table with type REALTIME", required = true, example = "myTable_REALTIME")
+      @PathParam("tableNameWithType") String tableNameWithType,
+      @ApiParam(value = "Name of the segment", required = true) @PathParam("segmentName") @Encoded String segmentName,
+      @Context HttpHeaders httpHeaders) {
+    tableNameWithType = DatabaseUtils.translateTableName(tableNameWithType, httpHeaders);
+    segmentName = URIUtils.decode(segmentName);
+    LOGGER.info("Received a request to get Crc for segment {} table {}", segmentName, tableNameWithType);
+    // Validate data access
+    ServerResourceUtils.validateDataAccess(_accessControlFactory, tableNameWithType, httpHeaders);
+
+    TableDataManager tableDataManager =
+        ServerResourceUtils.checkGetTableDataManager(_serverInstance, tableNameWithType);
+    SegmentDataManager segmentDataManager = tableDataManager.acquireSegment(segmentName);
+    if (segmentDataManager == null) {
+      throw new WebApplicationException(
+          String.format("Table %s segment %s does not exist", tableNameWithType, segmentName),
+          Response.Status.NOT_FOUND);
+    }
+    String crc = segmentDataManager.getSegment().getSegmentMetadata().getCrc();
+    tableDataManager.releaseSegment(segmentDataManager);
+    return crc;
+  }
+
   /**
    * Download snapshot for the given immutable segment for upsert table. This endpoint is used when get snapshot from
    * peer to avoid recompute when reload segments.