Merge upstream/main, resolving conflicts

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: pthirun

build.gradle

@@ -761,7 +761,6 @@ ext.createDiffFile = { ->
       // Keep this sorted
       // da-vinci-client
       ':!clients/da-vinci-client/src/main/java/com/linkedin/davinci/DaVinciBackend.java',
-      ':!clients/da-vinci-client/src/main/java/com/linkedin/davinci/ingestion/isolated/IsolatedIngestionServer.java',
       ':!clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/heartbeat/HeartbeatStatReporter.java',
 
       // venice-client

clients/da-vinci-client/build.gradle

@@ -45,6 +45,7 @@ dependencies {
   testImplementation project(':internal:venice-test-common')
   testImplementation project(':internal:venice-client-common').sourceSets.test.output
   testImplementation libraries.kafkaClientsTest
+  testImplementation libraries.classgraph
   testImplementation libraries.openTelemetryTestSdk
 }
 

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

@@ -20,23 +20,18 @@
 import com.linkedin.davinci.config.VeniceServerConfig;
 import com.linkedin.davinci.ingestion.DefaultIngestionBackend;
 import com.linkedin.davinci.ingestion.IngestionBackend;
-import com.linkedin.davinci.ingestion.IsolatedIngestionBackend;
-import com.linkedin.davinci.ingestion.main.MainIngestionStorageMetadataService;
-import com.linkedin.davinci.ingestion.utils.IsolatedIngestionUtils;
 import com.linkedin.davinci.kafka.consumer.KafkaStoreIngestionService;
 import com.linkedin.davinci.notifier.VeniceNotifier;
 import com.linkedin.davinci.repository.VeniceMetadataRepositoryBuilder;
 import com.linkedin.davinci.stats.AggBlobTransferStats;
 import com.linkedin.davinci.stats.AggVersionedBlobTransferStats;
 import com.linkedin.davinci.stats.AggVersionedStorageEngineStats;
 import com.linkedin.davinci.stats.HeartbeatMonitoringServiceStats;
-import com.linkedin.davinci.stats.MetadataUpdateStats;
 import com.linkedin.davinci.stats.RocksDBMemoryStats;
 import com.linkedin.davinci.stats.ingestion.heartbeat.HeartbeatMonitoringService;
 import com.linkedin.davinci.storage.StorageEngineMetadataService;
 import com.linkedin.davinci.storage.StorageMetadataService;
 import com.linkedin.davinci.storage.StorageService;
-import com.linkedin.davinci.store.StorageEngine;
 import com.linkedin.davinci.store.cache.backend.ObjectCacheBackend;
 import com.linkedin.davinci.store.cache.backend.ObjectCacheConfig;
 import com.linkedin.venice.annotation.VisibleForTesting;
@@ -51,7 +46,6 @@
 import com.linkedin.venice.kafka.protocol.state.PartitionState;
 import com.linkedin.venice.kafka.protocol.state.StoreVersionState;
 import com.linkedin.venice.meta.ClusterInfoProvider;
-import com.linkedin.venice.meta.IngestionMode;
 import com.linkedin.venice.meta.ReadOnlySchemaRepository;
 import com.linkedin.venice.meta.ReadOnlyStoreRepository;
 import com.linkedin.venice.meta.Store;
@@ -156,7 +150,7 @@ public DaVinciBackend(
       metricsRepository = Optional.ofNullable(clientConfig.getMetricsRepository())
           .orElse(TehutiUtils.getMetricsRepository(DAVINCI_CLIENT.getName()));
       VeniceMetadataRepositoryBuilder veniceMetadataRepositoryBuilder =
-          new VeniceMetadataRepositoryBuilder(configLoader, clientConfig, metricsRepository, icProvider, false);
+          new VeniceMetadataRepositoryBuilder(configLoader, clientConfig, metricsRepository, icProvider);
 
       ClusterInfoProvider clusterInfoProvider = veniceMetadataRepositoryBuilder.getClusterInfoProvider();
       ReadOnlyStoreRepository readOnlyStoreRepository = veniceMetadataRepositoryBuilder.getStoreRepo();
@@ -195,35 +189,17 @@ public DaVinciBackend(
               backendConfig.getRocksDBServerConfig().isRocksDBPlainTableFormatEnabled())
           : null;
 
-      // Add extra safeguards here to ensure we have released RocksDB database locks before we initialize storage
-      // services.
-      IsolatedIngestionUtils.destroyLingeringIsolatedIngestionProcess(configLoader);
       /**
        * The constructor of {@link #storageService} will take care of unused store/store version cleanup.
-       *
-       * When Ingestion Isolation is enabled, we don't want to restore data partitions here:
-       * 1. It is a waste of effort since all the opened partitioned will be closed right after.
-       * 2. When DaVinci memory limiter is enabled, currently, SSTFileManager doesn't clean up the entries belonging
-       *    to the closed database, which means when the Isolated process hands the database back to the main process,
-       *    some removed SST files (some SST files can be removed in Isolated Process because of log compaction) will
-       *    remain in SSTFileManager tracked file list.
-       * 3. We still want to open metadata partition, otherwise {@link StorageService} won't scan the local db folder.
-       *    Also opening metadata partition in main process won't cause much side effect from DaVinci memory limiter's
-       *    POV, since metadata partition won't be handed back to main process in the future.
-       * 4. When Ingestion Isolation is enabled with suppressing live update feature, main process needs to open all the
-       *    data partitions since Isolated Process won't re-ingest the existing partitions.
        */
-      boolean whetherToRestoreDataPartitions = !isIsolatedIngestion()
-          || configLoader.getVeniceServerConfig().freezeIngestionIfReadyToServeOrLocalDataExists();
-      LOGGER.info("DaVinci {} restore data partitions.", whetherToRestoreDataPartitions ? "will" : "won't");
       storageService = new StorageService(
           configLoader,
           aggVersionedStorageEngineStats,
           rocksDBMemoryStats,
           storeVersionStateSerializer,
           partitionStateSerializer,
           storeRepository,
-          whetherToRestoreDataPartitions,
+          true,
           true,
           functionToCheckWhetherStorageEngineShouldBeKeptOrNot(managedClients));
       storageService.start();
@@ -245,14 +221,8 @@ public DaVinciBackend(
         LOGGER.info("Successfully verified the latest protocols at runtime are valid in Venice backend.");
       }
 
-      storageMetadataService = backendConfig.getIngestionMode().equals(IngestionMode.ISOLATED)
-          ? new MainIngestionStorageMetadataService(
-              backendConfig.getIngestionServicePort(),
-              partitionStateSerializer,
-              new MetadataUpdateStats(metricsRepository),
-              configLoader,
-              storageService.getStoreVersionStateSyncer())
-          : new StorageEngineMetadataService(storageService.getStorageEngineRepository(), partitionStateSerializer);
+      storageMetadataService =
+          new StorageEngineMetadataService(storageService.getStorageEngineRepository(), partitionStateSerializer);
       // Start storage metadata service
       ((AbstractVeniceService) storageMetadataService).start();
       compressorFactory = new StorageEngineBackedCompressorFactory(storageMetadataService);
@@ -282,7 +252,6 @@ public DaVinciBackend(
           partitionStateSerializer,
           Optional.empty(),
           null,
-          false,
           compressorFactory,
           cacheBackend,
           true,
@@ -315,17 +284,7 @@ public DaVinciBackend(
 
       ingestionService.start();
 
-      if (isIsolatedIngestion() && cacheConfig.isPresent()) {
-        // TODO: There are 'some' cases where this mix might be ok, (like a batch only store, or with certain TTL
-        // settings),
-        // could add further validation. If the process isn't ingesting data, then it can't maintain the object cache
-        // with
-        // a correct view of the data.
-        throw new IllegalArgumentException(
-            "Ingestion isolated and Cache are incompatible configs!!  Aborting start up!");
-      }
-
-      if (BlobTransferUtils.isBlobTransferManagerEnabled(backendConfig, isIsolatedIngestion())) {
+      if (BlobTransferUtils.isBlobTransferManagerEnabled(backendConfig)) {
         aggVersionedBlobTransferStats =
             new AggVersionedBlobTransferStats(metricsRepository, storeRepository, configLoader.getVeniceServerConfig());
         aggBlobTransferStats =
@@ -441,41 +400,12 @@ private Function<String, Boolean> functionToCheckWhetherStorageEngineShouldBeKep
 
   @VisibleForTesting
   final synchronized void bootstrap() {
-    /**
-     * In order to make bootstrap logic compatible with ingestion isolation, we first scan all local storage engines,
-     * record all store versions that are up-to-date and close all storage engines. This will make sure child process
-     * can open RocksDB stores.
-     */
-    if (isIsolatedIngestion()) {
-      if (configLoader.getVeniceServerConfig().freezeIngestionIfReadyToServeOrLocalDataExists()) {
-        /**
-         * In this case we will only need to close metadata partition, as it is supposed to be opened and managed by
-         * forked ingestion process via following subscribe call.
-         */
-        for (StorageEngine storageEngine: getStorageService().getStorageEngineRepository()
-            .getAllLocalStorageEngines()) {
-          storageEngine.closeMetadataPartition();
-        }
-      } else {
-        getStorageService().closeAllStorageEngines();
-      }
-    }
-
-    ingestionBackend = isIsolatedIngestion()
-        ? new IsolatedIngestionBackend(
-            configLoader,
-            metricsRepository,
-            storageMetadataService,
-            ingestionService,
-            getStorageService(),
-            blobTransferManager,
-            this::getVeniceCurrentVersionNumber)
-        : new DefaultIngestionBackend(
-            storageMetadataService,
-            ingestionService,
-            getStorageService(),
-            blobTransferManager,
-            configLoader.getVeniceServerConfig());
+    ingestionBackend = new DefaultIngestionBackend(
+        storageMetadataService,
+        ingestionService,
+        getStorageService(),
+        blobTransferManager,
+        configLoader.getVeniceServerConfig());
     ingestionBackend.addIngestionNotifier(ingestionListener);
   }
 
@@ -636,10 +566,6 @@ protected void deleteStore(String storeName) {
     }
   }
 
-  public final boolean isIsolatedIngestion() {
-    return configLoader.getVeniceServerConfig().getIngestionMode().equals(IngestionMode.ISOLATED);
-  }
-
   // Move the logic to this protected method to make it visible for unit test.
   protected void handleStoreChanged(StoreBackend storeBackend) {
     // Skip version swaps for version-specific stores

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

@@ -17,7 +17,6 @@
 import com.linkedin.venice.compute.ComputeRequestWrapper;
 import com.linkedin.venice.compute.ComputeUtils;
 import com.linkedin.venice.exceptions.VeniceException;
-import com.linkedin.venice.meta.IngestionMode;
 import com.linkedin.venice.meta.Store;
 import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.partitioner.VenicePartitioner;
@@ -93,14 +92,6 @@ public class VersionBackend {
     this.config = backend.getConfigLoader().getStoreConfig(version.kafkaTopicName());
     this.batchReportEOIPStatusEnabled = config.getBatchReportEOIPEnabled();
 
-    if (this.config.getIngestionMode().equals(IngestionMode.ISOLATED)) {
-      /*
-       * Explicitly disable the store restore since we don't want to open other partitions that should be controlled by
-       * child process. All the finished partitions will be closed by child process and reopened in parent process.
-       */
-      this.config.setRestoreDataPartitions(false);
-      this.config.setRestoreMetadataPartition(false);
-    }
     this.partitioner = PartitionUtils.getUserPartitionLevelVenicePartitioner(version.getPartitionerConfig());
     this.suppressLiveUpdates = this.config.freezeIngestionIfReadyToServeOrLocalDataExists();
     this.storageEngine.set(backend.getStorageService().getStorageEngine(version.kafkaTopicName()));
@@ -416,7 +407,8 @@ synchronized CompletableFuture<Void> subscribe(
                   partition,
                   checkpointInfo.getTimestampsMap(),
                   checkpointInfo.getPostitionMap());
-      backend.getIngestionBackend().startConsumption(config, partition, pubSubPosition);
+      String replicaId = Utils.getReplicaId(version.kafkaTopicName(), partition);
+      backend.getIngestionBackend().startConsumption(config, partition, pubSubPosition, replicaId);
       tryStartHeartbeat();
     }
 
@@ -473,7 +465,9 @@ synchronized void unsubscribe(ComplementSet<Integer> partitions) {
       completePartition(partition);
       backend.getHeartbeatMonitoringService()
           .updateLagMonitor(version.kafkaTopicName(), partition, HeartbeatLagMonitorAction.REMOVE_MONITOR);
-      backend.getIngestionBackend().dropStoragePartitionGracefully(config, partition, stopConsumptionTimeoutInSeconds);
+      String replicaId = Utils.getReplicaId(version.kafkaTopicName(), partition);
+      backend.getIngestionBackend()
+          .dropStoragePartitionGracefully(config, partition, stopConsumptionTimeoutInSeconds, replicaId);
       partitionFutures.remove(partition);
       partitionToPendingReportIncrementalPushList.remove(partition);
       partitionToBatchReportEOIPEnabled.remove(partition);

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

@@ -218,20 +218,12 @@ private static boolean isBlobTransferAclValidationEnabled(VeniceConfigLoader con
   /**
    * A config check to determine if blob transfer manager is enabled
    * @param backendConfig the Venice server config
-   * @param isIsolatedIngestionEnabled whether isolated ingestion is enabled
    * @return true if blob transfer manager is enabled, false otherwise
    */
-  public static boolean isBlobTransferManagerEnabled(
-      VeniceServerConfig backendConfig,
-      boolean isIsolatedIngestionEnabled) {
-    // Blob transfer feature and isolated ingestion feature are mutually exclusive
+  public static boolean isBlobTransferManagerEnabled(VeniceServerConfig backendConfig) {
     if (backendConfig.isBlobTransferManagerEnabled() && backendConfig.isBlobTransferSslEnabled()
         && backendConfig.isBlobTransferAclEnabled()) {
-      if (isIsolatedIngestionEnabled) {
-        throw new VeniceException("Blob transfer manager is not supported with isolated ingestion");
-      } else {
-        return true;
-      }
+      return true;
     } else if (backendConfig.isBlobTransferManagerEnabled()) {
       throw new VeniceException("Blob transfer manager is not supported without SSL and ACL enabled");
     }

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

@@ -277,27 +277,18 @@ private Optional<Version> getVersion() {
   }
 
   protected CompletableFuture<Void> seekToTail() {
-    if (getBackend().isIsolatedIngestion()) {
-      throw new VeniceClientException("Isolated Ingestion is not supported with seekToCheckpoint");
-    }
     throwIfNotReady();
     addPartitionsToSubscription(ComplementSet.universalSet());
     return getStoreBackend().seekToCheckpoint(new DaVinciSeekCheckpointInfo(null, null, null, true), getVersion());
   }
 
   protected CompletableFuture<Void> seekToTail(Set<Integer> partitionSet) {
-    if (getBackend().isIsolatedIngestion()) {
-      throw new VeniceClientException("Isolated Ingestion is not supported with seekToCheckpoint");
-    }
     throwIfNotReady();
     addPartitionsToSubscription(ComplementSet.wrap(partitionSet));
     return getStoreBackend().seekToCheckpoint(new DaVinciSeekCheckpointInfo(null, null, null, true), getVersion());
   }
 
   protected CompletableFuture<Void> seekToCheckpoint(Set<VeniceChangeCoordinate> checkpoints) {
-    if (getBackend().isIsolatedIngestion()) {
-      throw new VeniceClientException("Isolated Ingestion is not supported with seekToCheckpoint");
-    }
     throwIfNotReady();
     Map<Integer, PubSubPosition> positionMap = new HashMap<>();
     for (VeniceChangeCoordinate changeCoordinate: checkpoints) {
@@ -313,19 +304,13 @@ protected CompletableFuture<Void> seekToCheckpoint(Set<VeniceChangeCoordinate> c
   }
 
   protected CompletableFuture<Void> seekToTimestamps(Map<Integer, Long> timestamps) {
-    if (getBackend().isIsolatedIngestion()) {
-      throw new VeniceClientException("Isolated Ingestion is not supported with seekToTimestamps");
-    }
     throwIfNotReady();
     addPartitionsToSubscription(ComplementSet.wrap(timestamps.keySet()));
     return getStoreBackend()
         .seekToCheckpoint(new DaVinciSeekCheckpointInfo(null, timestamps, null, false), getVersion());
   }
 
   protected CompletableFuture<Void> seekToTimestamps(Long timestamp) {
-    if (getBackend().isIsolatedIngestion()) {
-      throw new VeniceClientException("Isolated Ingestion is not supported with seekToTimestamps");
-    }
     throwIfNotReady();
     addPartitionsToSubscription(ComplementSet.universalSet());
     return getStoreBackend()