[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.

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

@@ -154,7 +154,13 @@ synchronized CompletableFuture<Void> subscribe(
       // Recreate store config that was potentially deleted by unsubscribe.
       config.store();
     }
+
+    ComplementSet<Integer> unassignedPartitionSet = ComplementSet.newSet(subscription);
+    // Set of partitions that are not included in the submitted set of partitions.
+    unassignedPartitionSet.removeAll(partitions);
+
     subscription.addAll(partitions);
+    unsubscribe(unassignedPartitionSet);
 
     if (daVinciFutureVersion == null) {
       trySubscribeDaVinciFutureVersion();