[da-vinci][dvc] Dropping unassigned partitions

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

@@ -207,6 +207,7 @@ public DaVinciBackend(
       String pid = Utils.getPid();
       String instanceSuffix =
           configLoader.getCombinedProperties().getString(PUSH_STATUS_INSTANCE_NAME_SUFFIX, (pid == null ? "NA" : pid));
+      // Current instance name.
       String instanceName = Utils.getHostName() + "_" + instanceSuffix;
 
       // Fetch latest update schema's protocol ID for Push Status Store from Router.
@@ -474,6 +475,11 @@ private synchronized void bootstrap() {
       AbstractStorageEngine storageEngine = storageService.getStorageEngine(versionTopic);
       aggVersionedStorageEngineStats.setStorageEngine(versionTopic, storageEngine);
       StoreBackend storeBackend = getStoreOrThrow(storeName);
+      ComplementSet<Integer> subscription = storeBackend.getSubscription();
+      ComplementSet<Integer> unassignedPartitionSet = ComplementSet.newSet(subscription);
+      // Set of partitions that are not included in the submitted set of partitions.
+      unassignedPartitionSet.removeAll(ComplementSet.newSet(partitions));
+      subscription.removeAll(unassignedPartitionSet);
       storeBackend.subscribe(ComplementSet.newSet(partitions), Optional.of(version));
     });
   }

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

@@ -108,6 +108,10 @@ public StoreBackendStats getStats() {
     return stats;
   }
 
+  public ComplementSet<Integer> getSubscription() {
+    return subscription;
+  }
+
   public ReferenceCounted<VersionBackend> getDaVinciCurrentVersion() {
     return daVinciCurrentVersionRef.get();
   }
@@ -154,6 +158,7 @@ synchronized CompletableFuture<Void> subscribe(
       // Recreate store config that was potentially deleted by unsubscribe.
       config.store();
     }
+
     subscription.addAll(partitions);
 
     if (daVinciFutureVersion == null) {