remove consuming segment from PartitionUpsertMetadataManager to prevent memory leak

[richardstartin]

Motivated by a heap dump from an OSS user where their realtime servers were OOM'd with a 16GB _primaryKeyToRecordLocationMap

Class Name	Shallow Heap	Retained Heap
java.util.concurrent.ConcurrentHashMap$Node[134217728] @ 0x727800000	536,870,928	16,829,827,816
table java.util.concurrent.ConcurrentHashMap @ 0x40343ec78	64	16,829,827,880
_primaryKeyToRecordLocationMap org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager @ 0x40343ec48	48	16,829,835,136

[codecov-commenter]

Codecov Report

Merging #8034 (3e915e5) into master (22d583c) will decrease coverage by 5.65%.
The diff coverage is 65.78%.

❗ Current head 3e915e5 differs from pull request most recent head 17a13b2. Consider uploading reports for the commit 17a13b2 to get more accurate results

@@             Coverage Diff              @@
##             master    #8034      +/-   ##
============================================
- Coverage     70.33%   64.68%   -5.66%     
- Complexity     4215     4261      +46     
============================================
  Files          1596     1560      -36     
  Lines         82796    81516    -1280     
  Branches      12354    12252     -102     
============================================
- Hits          58234    52728    -5506     
- Misses        20579    25032    +4453     
+ Partials       3983     3756     -227     

Flag	Coverage Δ
integration1	?
unittests1	67.87% <68.70%> (-0.25%)	⬇️
unittests2	14.26% <6.31%> (-0.10%)	⬇️

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

Impacted Files	Coverage Δ
...t/broker/api/resources/PinotBrokerHealthCheck.java	0.00% <0.00%> (ø)
...e/pinot/common/utils/FileUploadDownloadClient.java	16.92% <0.00%> (-40.39%)	⬇️
...a/org/apache/pinot/common/utils/ServiceStatus.java	50.95% <0.00%> (-11.30%)	⬇️
...apache/pinot/common/utils/SqlResultComparator.java	0.00% <ø> (ø)
...pinot/common/utils/fetcher/BaseSegmentFetcher.java	60.60% <0.00%> (+9.09%)	⬆️
...ller/api/resources/PinotControllerHealthCheck.java	0.00% <0.00%> (ø)
.../controller/helix/ControllerRequestURLBuilder.java	74.01% <0.00%> (-9.32%)	⬇️
...ntroller/helix/core/PinotHelixResourceManager.java	61.67% <ø> (-3.77%)	⬇️
...controller/helix/core/minion/PinotTaskManager.java	41.98% <0.00%> (-23.54%)	⬇️
...org/apache/pinot/core/auth/BasicAuthPrincipal.java	72.72% <0.00%> (+12.72%)	⬆️
... and 393 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 22d583c...17a13b2. Read the comment docs.

[Jackie-Jiang]

This shouldn't be required because when the segment is sealed and loaded again as ImmutableSegment, all the entries in the upsert metadata map should have already been replaced. This leak might happen only when user explicitly delete the consuming segment, but that will stall the realtime consumption and also mess up the upsert metadata, so I don't think we need to handle this scenario.

[richardstartin]

This shouldn't be required because when the segment is sealed and loaded again as ImmutableSegment, all the entries in the upsert metadata map should have already been replaced. This leak might happen only when user explicitly delete the consuming segment, but that will stall the realtime consumption and also mess up the upsert metadata, so I don't think we need to handle this scenario.

What about when it's loaded on a different server?

[Jackie-Jiang]

This shouldn't be required because when the segment is sealed and loaded again as ImmutableSegment, all the entries in the upsert metadata map should have already been replaced. This leak might happen only when user explicitly delete the consuming segment, but that will stall the realtime consumption and also mess up the upsert metadata, so I don't think we need to handle this scenario.

What about when it's loaded on a different server?

It shouldn't be possible as the state transition always go from CONSUMING to ONLINE on the same server. Note that only the consuming segment is MutableSegment, the sealed one will become ImmutableSegment.