[fix][ml] OOM caused by orphan items in ActiveManagedCursorContainerImpl.pendingPositionUpdates - #26506

Motivation

The issue was introduced by #24623, only affects the version newer than 4.1.0

Background

• ActiveManagedCursorContainerImpl traces the slowest position of cursors.
• To delay the calculation of the slowest position, ActiveManagedCursorContainerImpl maintains two collections: pendingRemovedCursors and pendingPositionUpdates.
• pendingRemovedCursors caches the events that cursors are inactive, and pendingPositionUpdates caches the events that cursors are active.
• When getting the slowest position, it flushes pendingRemovedCursors and pendingPositionUpdates.
  • (Highlight) If no such call was made, it never flushes

Issue

With cacheEvictionByExpectedReadCount enabled, a cursor can be removed before its read any messages. The removed cursor is no longer tracked in the Managed ledger, but its node remains in ActiveManagedCursorContainerImpl.pendingPositionUpdates until a future pending-position flush. If no such flush is triggered, repeated cursor creation and removal can accumulate stale nodes and retain their associated cursor objects, increasing memory usage and potentially causing OOM.

It can be reproduced with the following steps. See also the new test.

• create a topic
• create subscription 1
• delete the subscription
  • an orphan item is left in ActiveManagedCursorContainerImpl.pendingPositionUpdates.
• create subscription 2
• delete the subscription
  • the second orphan item is left in ActiveManagedCursorContainerImpl.pendingPositionUpdates.
• ...

Modifications

Count orphan items in ActiveManagedCursorContainerImpl.pendingPositionUpdates, flush pending cursor updates, and reset the counter when the count of orphan items is 10.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment