[server][dvc] Do not drop PCS from partitionConsumptionStateMap for non-Helix triggered actions

Remove PCS if it's dvc client

Fix flaky test

Fix another flaky test

Address review comments

Delete test that's already covered in other tests

Redistributed tests

Author: sushantmane

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java

@@ -2368,11 +2368,12 @@ protected void processCommonConsumerAction(ConsumerAction consumerAction) throws
          * {@link #kafkaDataValidationService}, we would like to drain all the buffered messages before cleaning up those
          * two variables to avoid the race condition.
          */
-        partitionConsumptionStateMap.remove(partition);
-        if (consumerAction.isHelixTriggeredAction()) {
+        if (isDaVinciClient() || consumerAction.isHelixTriggeredAction()) {
           LOGGER.info(
-              "Removing tracking of replica: {} from storage utilization manager as this UNSUBSCRIBE is helix triggered action",
-              topicPartition);
+              "Removing replica: {} from PCS map and storage utilization manager. Trigger: {}",
+              topicPartition,
+              isDaVinciClient() ? "DaVinci client unsubscribe" : "Helix-triggered unsubscribe");
+          partitionConsumptionStateMap.remove(partition);
           storageUtilizationManager.removePartition(partition);
         }
         getDataIntegrityValidator().clearPartition(partition);