[server][dvc] Preserve PCS during non-Helix UNSUBSCRIBE to maintain quota accuracy

[sushantmane]

[server][dvc] Preserve PCS during non-Helix UNSUBSCRIBE to maintain quota accuracy

Keep PartitionConsumptionState entries in partitionConsumptionStateMap
for non-Helix UNSUBSCRIBE actions. Previously PCS was removed for all
UNSUBSCRIBE triggers, causing quota computation to lose visibility into
partitions unsubscribed due to errors or after end-of-push for batch stores.

PCS is now removed only for Helix-initiated or for DVC UNSUBSCRIBE, to
ensure storage utilization tracking and quota enforcement remain accurate.

Code changes

• Added new code behind a config. If so list the config names and their default values in the PR description.
• Introduced new log lines.
  • Confirmed if logs need to be rate limited to avoid excessive logging.

Concurrency-Specific Checks

Both reviewer and PR author to verify

• Code has no race conditions or thread safety issues.
• Proper synchronization mechanisms (e.g., synchronized, RWLock) are used where needed.
• No blocking calls inside critical sections that could lead to deadlocks or performance degradation.
• Verified thread-safe collections are used (e.g., ConcurrentHashMap, CopyOnWriteArrayList).
• Validated proper exception handling in multi-threaded code to avoid silent thread termination.

How was this PR tested?

• New unit tests added.
• New integration tests added.
• Modified or extended existing tests.
• Verified backward compatibility (if applicable).

Does this PR introduce any user-facing or breaking changes?

• No. You can skip the rest of this section.
• Yes. Clearly explain the behavior change and its impact.

[Copilot]

Pull Request Overview

This PR fixes a quota tracking issue by preserving PartitionConsumptionState (PCS) entries during non-Helix UNSUBSCRIBE operations. Previously, PCS was removed for all unsubscribe triggers, causing the quota computation to lose visibility into partitions that were unsubscribed due to errors or after end-of-push for batch stores.

• Moved partitionConsumptionStateMap.remove(partition) inside the Helix-triggered action conditional
• PCS now only removed for Helix-initiated unsubscribe, ensuring storage utilization tracking and quota enforcement remain accurate
• Aligns with existing pattern for failedPartitions and storageUtilizationManager cleanup

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated 1 comment.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review for a chance to win a $100 gift card. Take the survey.

[Copilot]

The log message has a logical inconsistency. When isDaVinciClient() returns true, the ternary operator will always select "DaVinci client unsubscribe", even if the action is also Helix-triggered. This could be misleading in cases where a DaVinci client uses Helix-triggered unsubscribe.

Consider restructuring the log message to accurately reflect both conditions:

String trigger = isDaVinciClient() 
    ? (consumerAction.isHelixTriggeredAction() ? "DaVinci client (Helix-triggered)" : "DaVinci client") 
    : "Helix-triggered unsubscribe";
LOGGER.info(
    "Removing replica: {} from PCS map and storage utilization manager. Trigger: {}",
    topicPartition,
    trigger);