[da-vinci][server] Move blob transfer orchestration into StoreIngestion Task with SIT-driven completion (linkedin#2601)

Author: jingy-li

clients/da-vinci-client/src/main/java/com/linkedin/davinci/DaVinciBackend.java

@@ -329,6 +329,11 @@ public DaVinciBackend(
         blobTransferManager = null;
       }
 
+      // Inject blob transfer manager into ingestion service so SIT can use it
+      if (blobTransferManager != null) {
+        ingestionService.setBlobTransferManager(blobTransferManager);
+      }
+
       bootstrap();
 
       storeRepository.registerStoreDataChangedListener(storeChangeListener);
@@ -410,7 +415,6 @@ final synchronized void bootstrap() {
         storageMetadataService,
         ingestionService,
         getStorageService(),
-        blobTransferManager,
         configLoader.getVeniceServerConfig());
     ingestionBackend.addIngestionNotifier(ingestionListener);
   }

clients/da-vinci-client/src/main/java/com/linkedin/davinci/helix/HelixParticipationService.java

@@ -130,11 +130,15 @@ public HelixParticipationService(
       throw new VeniceException("Expecting " + KafkaStoreIngestionService.class.getName() + " for ingestion backend!");
     }
 
+    // Inject blob transfer manager into ingestion service so SIT can use it
+    if (blobTransferManager != null) {
+      ((KafkaStoreIngestionService) storeIngestionService).setBlobTransferManager(blobTransferManager);
+    }
+
     this.ingestionBackend = new DefaultIngestionBackend(
         storageMetadataService,
         (KafkaStoreIngestionService) storeIngestionService,
         storageService,
-        blobTransferManager,
         veniceConfigLoader.getVeniceServerConfig());
   }