[Subscription] Fix pending consensus request memory leak

[Caideyipi]

Description

Bound pending consensus request memory

In a multi-replica region, LogDispatcher serializes each IndexedConsensusRequest before the same object is offered to subscription pending queues. A slow or paused subscription could therefore retain both the raw InsertNode and serialized ByteBuffer without participating in the IoTConsensus memory budget, causing heap growth and eventually DataNode fencing.

This PR:

• makes request serialization idempotent and exposes the retained size as raw request memory plus serialized buffer capacity;
• reuses the existing ref-counted IoTConsensusMemoryManager, so dispatchers and subscription queues sharing one request reserve its physical memory only once;
• rejects subscription fast-path admission when the shared queue budget is exhausted, allowing the existing WAL replay path to recover the entry;
• releases queue ownership after materialization in a finally block, and on offer failure, clear, deactivation, and close;
• keeps batch accounting unchanged because replication batches no longer own the raw request objects.

Remove unnecessary retention

Remove IoTConsensusServerImpl.lastConsensusRequest, which kept the latest request, including a potentially large raw/serialized payload, reachable indefinitely.

Diagnostics

Expose retainedRequestBytes in the subscription queue core report to make pending-request retention observable.

Tests

• IoTConsensusMemoryManagerTest: 4 tests passed.
• ConsensusPrefetchingQueueTest, ConsensusPrefetchingQueueDataNodeMemoryTest, and ConsensusPrefetchingQueueWalBackpressureTest: 29 tests passed.
• Spotless, Checkstyle, and git diff --check passed.

---

This PR has:

• been self-reviewed.
  • concurrent write
• added Javadocs for non-trivial memory accounting behavior.
• added comments explaining ownership and release intent.
• added or updated unit tests to cover new code paths.

---

Key changed/added classes

• IndexedConsensusRequest
• IoTConsensusMemoryManager
• IoTConsensusServerImpl
• ConsensusPrefetchingQueue

[jt2594838]

This is for debugging purposes, where the request is stuck or fails to be found in the WAL.