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

Author: sushantmane

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

@@ -2369,8 +2369,8 @@ 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()) {
+          partitionConsumptionStateMap.remove(partition);
           LOGGER.info(
               "Removing tracking of replica: {} from storage utilization manager as this UNSUBSCRIBE is helix triggered action",
               topicPartition);