Eliminate duplicate cancel attempts in PerQueryCPUMemAccountant

[vrajat]

In critical mode, the accountant sends a cancel to the most expensive query. This query may not terminate immediately. So the accountant will attempt multiple times to cancel the query. A sinister side effect is that it also increments _numQueriesKilledConsecutively and calls GC when the value is above a config. Note that the GC call is useless as the query hasnt cancelled yet and released the memory allocations.

The accountant now remembers the queries that have been cancelled in _cancelSentQueries. It finds the most expensive query which hasn't been cancelled.

          maxUsageTuple = _aggregatedUsagePerActiveQuery.values().stream()
              .filter(stats -> !_cancelSentQueries.contains(stats.getQueryId()))
              .max(Comparator.comparing(AggregatedStats::getAllocatedBytes))
              .orElse(null);

There are a few more changes that was required to safely add this feature.
PerQueryCPUMemAccountantFactory.aggregate(boolean triggered) performs two functions:

• It reaps finished tasks and associated metadata - this is relevant to this feature. _cancelledSentQueries entries also have to be reaped.
• If triggered, it also aggregates resource usage from _threadEntriesMap to a Map<String, AggregatedStats>.

The dual functionality made it hard to write unit tests. So the function has been split into:

• reapFinishedTasks
• getQueryResources - this already existed.
  The main change is that if (isTriggered) { ... } blocks have been removed.

Another change to help with testing is that PerQueryCPUMemAccountant.WatcherTask.run() has been extracted to PerQueryCPUMemAccountant.WatcherTask.runOnce(). This helps to run one iteration at a time in tests. Also it is possible to override this function in a test to only focus on relevant operations.

There are other minor changes to change member visibility to protected and new getter functions to also help with testing.

Consequently, the other major contribution of this PR is unit tests for reap/aggregate functions as well as testing for duplicate cancel attempts.

Closes #16110

[Copilot]

Pull Request Overview

This PR refactors the PerQueryCPUMemAccountant to remember which queries have already been sent cancel operations (preventing duplicate cancels), splits its aggregate logic into two clearer methods, and adds targeted unit tests to verify resource aggregation and cancel behavior.

• Introduce and maintain a _cancelSentQueries set to filter out already-cancelled queries
• Split the old aggregate(boolean) method into reapFinishedTasks() (moves finished-task data) and getQueryResourcesImpl() (collects active-resource stats)
• Extract WatcherTask.runOnce(), adjust member visibility, and add getters to make testing easier
• Add comprehensive unit tests covering aggregation in various thread/task states and cancel de-duplication

Reviewed Changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
pinot-core/src/test/java/org/apache/pinot/core/accounting/TestResourceAccountant.java	New test helper that seeds thread entries for resource-accountant tests
pinot-core/src/test/java/org/apache/pinot/core/accounting/QueryMonitorConfigTest.java	Renamed test class to match file name and purpose
pinot-core/src/test/java/org/apache/pinot/core/accounting/PerQueryCPUMemAccountantTest.java	Added unit tests for resource aggregation across live, new, and finished tasks
pinot-core/src/test/java/org/apache/pinot/core/accounting/PerQueryCPUMemAccountCancelTest.java	Added unit tests verifying that duplicate cancel attempts are suppressed
pinot-core/src/main/java/org/apache/pinot/core/accounting/PerQueryCPUMemAccountantFactory.java	Core refactor: track cancelled queries, refactor aggregation flow, extract runOnce

Comments suppressed due to low confidence (1)

pinot-core/src/test/java/org/apache/pinot/core/accounting/PerQueryCPUMemAccountantTest.java:183

• [nitpick] Test method name testInActiveQuerySet uses inconsistent camel casing. Rename to testInactiveQuerySet for clarity and style consistency.

  void testInActiveQuerySet() {

[Copilot]

Removing entries from _threadEntriesMap inside a for-each over its own entrySet() can trigger a ConcurrentModificationException. Use an iterator and call iterator.remove() instead to safely remove dead threads.

Suggested change

	_threadEntriesMap.remove(thread);
	iterator.remove();

[vrajat]

cc @praveenc7

[codecov-commenter]

Codecov Report

❌ Patch coverage is 75.29412% with 21 lines in your changes missing coverage. Please review.
✅ Project coverage is 63.13%. Comparing base (1a476de) to head (624f30a).
⚠️ Report is 967 commits behind head on master.

Files with missing lines	Patch %	Lines
...re/accounting/PerQueryCPUMemAccountantFactory.java	75.29%	10 Missing and 11 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #16299      +/-   ##
============================================
+ Coverage     62.90%   63.13%   +0.23%     
+ Complexity     1386     1363      -23     
============================================
  Files          2867     2973     +106     
  Lines        163354   172530    +9176     
  Branches      24952    26428    +1476     
============================================
+ Hits         102755   108927    +6172     
- Misses        52847    55271    +2424     
- Partials       7752     8332     +580     

Flag	Coverage Δ
custom-integration1	100.00% <ø> (ø)
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-11	63.11% <75.29%> (+0.24%)	⬆️
java-21	63.10% <75.29%> (+0.28%)	⬆️
skip-bytebuffers-false	?
skip-bytebuffers-true	?
temurin	63.13% <75.29%> (+0.23%)	⬆️
unittests	63.13% <75.29%> (+0.23%)	⬆️
unittests1	56.34% <75.29%> (+0.52%)	⬆️
unittests2	33.19% <0.00%> (-0.38%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.