[da-vinci] Remove deprecated Ingestion Isolation feature

Remove the Ingestion Isolation feature which enabled running ingestion
in a separate forked JVM process communicating via HTTP/Netty IPC. This
feature is no longer needed, and removing it reduces codebase complexity
and maintenance burden.

- Delete ~30 production and test files: IsolatedIngestionBackend,
  IsolatedIngestionServer, MainIngestionMonitorService, and related
  classes in ingestion/isolated/, ingestion/main/, and utils/
- Delete IngestionMode and IngestionAction enums
- Remove all SERVER_INGESTION_ISOLATION_* config keys from ConfigKeys
- Simplify DaVinciBackend to always use DefaultIngestionBackend and
  StorageEngineMetadataService
- Remove isIsolatedIngestion parameter from StoreIngestionTask,
  KafkaStoreIngestionService, BlobTransferUtils, and
  VeniceMetadataRepositoryBuilder
- Remove ingestion isolation guards from DefaultIngestionBackend,
  AvroGenericDaVinciClient, and VersionBackend
- Clean up SpotBugs exclusions, server properties, and build configs
- Update integration tests to remove isolation-specific parameterization

Co-Authored-By: Claude Opus 4.6 <[email protected]>

Author: sixpluszero

build.gradle

@@ -755,7 +755,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/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()));

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()

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

@@ -131,9 +131,6 @@
 import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_CHECKPOINT_DURING_GRACEFUL_SHUTDOWN_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_HEARTBEAT_INTERVAL_MS;
 import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_INFO_LOG_LINE_LIMIT;
-import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_ISOLATION_APPLICATION_PORT;
-import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_ISOLATION_SERVICE_PORT;
-import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_MODE;
 import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_TASK_MAX_IDLE_COUNT;
 import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_TASK_REUSABLE_OBJECTS_STRATEGY;
 import static com.linkedin.venice.ConfigKeys.SERVER_KAFKA_CONSUMER_OFFSET_COLLECTION_ENABLED;
@@ -249,7 +246,6 @@
 import com.linkedin.venice.authorization.DefaultIdentityParser;
 import com.linkedin.venice.exceptions.ConfigurationException;
 import com.linkedin.venice.exceptions.VeniceException;
-import com.linkedin.venice.meta.IngestionMode;
 import com.linkedin.venice.pubsub.PubSubClientsFactory;
 import com.linkedin.venice.throttle.VeniceRateLimiter;
 import com.linkedin.venice.utils.ConfigCommonUtils;
@@ -275,7 +271,7 @@
 
 
 /**
- * VeniceServerConfig maintains configs specific to Venice Server, Da Vinci client and Isolated Ingestion Service.
+ * VeniceServerConfig maintains configs specific to Venice Server, Da Vinci client.
  */
 public class VeniceServerConfig extends VeniceClusterConfig {
   private static final Logger LOGGER = LogManager.getLogger(VeniceServerConfig.class);
@@ -496,9 +492,6 @@ public class VeniceServerConfig extends VeniceClusterConfig {
   private final int consumerPoolSizePerKafkaCluster;
   private final boolean leakedResourceCleanupEnabled;
 
-  private final IngestionMode ingestionMode;
-  private final int ingestionServicePort;
-  private final int ingestionApplicationPort;
   private final boolean databaseChecksumVerificationEnabled;
   private final boolean rocksDbStorageEngineConfigCheckEnabled;
 
@@ -923,10 +916,6 @@ public VeniceServerConfig(VeniceProperties serverProperties, Map<String, Map<Str
     }
     leakedResourceCleanupEnabled = serverProperties.getBoolean(SERVER_LEAKED_RESOURCE_CLEANUP_ENABLED, true);
 
-    ingestionMode =
-        IngestionMode.valueOf(serverProperties.getString(SERVER_INGESTION_MODE, IngestionMode.BUILT_IN.toString()));
-    ingestionServicePort = serverProperties.getInt(SERVER_INGESTION_ISOLATION_SERVICE_PORT, 27015);
-    ingestionApplicationPort = serverProperties.getInt(SERVER_INGESTION_ISOLATION_APPLICATION_PORT, 27016);
     databaseChecksumVerificationEnabled =
         serverProperties.getBoolean(SERVER_DATABASE_CHECKSUM_VERIFICATION_ENABLED, false);
 
@@ -1543,18 +1532,6 @@ public boolean isLeakedResourceCleanupEnabled() {
     return leakedResourceCleanupEnabled;
   }
 
-  public IngestionMode getIngestionMode() {
-    return ingestionMode;
-  }
-
-  public int getIngestionServicePort() {
-    return ingestionServicePort;
-  }
-
-  public int getIngestionApplicationPort() {
-    return ingestionApplicationPort;
-  }
-
   public boolean isDatabaseChecksumVerificationEnabled() {
     return databaseChecksumVerificationEnabled;
   }