[Subscription] Prevent event loss on poll payload overflow

[Caideyipi]

Description

Root cause

The subscription brokers polled events before enforcing the cumulative response payload limit. When several individually valid events were selected for one response, the receiver could discover only during serialization that the cumulative payload exceeded the poll budget. It then nacked the event. After 10 identical retries, poison-message handling force-acked the event and permanently skipped its rows.

This matches regression test case 1392 exactly: two skipped writer_B events, each containing 10 rows, account for the 20-row gap between benchmark/IoTDB count (42,423,170) and consumed count (42,423,150).

The failing event itself was about 30 MB, not 68 MB. The reported 68,135,206 bytes was the cumulative response size after adding it. The effective server threshold was 60,397,977 bytes, which is 90% of the consumer's 64 MiB poll budget. Therefore the write path's payload limit was not violated; whether a writer_B event succeeded depended on how much data had already been selected for that poll response.

Fix

• Check the cumulative payload budget before adding an event in both consensus and pipe subscription brokers.
• Return an event that does not fit to its prefetch queue atomically, without calling nack() or increasing nackCount.
• Apply the same budget enforcement across multiple brokers in SubscriptionBrokerAgent.
• Keep receiver-side overflow handling as a race-condition fallback: requeue cumulative-overflow events and nack only when requeue is not applicable or fails.
• Preserve existing behavior for a single event that is larger than the complete response budget, avoiding an infinite requeue loop.

Verification

• mvn spotless:apply -pl iotdb-core/datanode
• mvn -o test-compile -pl iotdb-core/datanode -DskipTests
• ConsensusSubscriptionBrokerPayloadLimitTest: 1 test, 0 failures/errors
• SubscriptionBrokerAgentPayloadLimitTest: 1 test, 0 failures/errors
• ConsensusPrefetchingQueueTest: 27 tests, 0 failures/errors
• git diff --check

---

This PR has:

• been self-reviewed.
  • concurrent read
  • concurrent write
  • concurrent read and write
• added Javadocs for non-trivial methods.
• added comments explaining the intent of payload-overflow requeue behavior.
• added or updated unit tests for the new code paths.

---

Key changed/added classes

• SubscriptionBrokerAgent
• ConsensusSubscriptionBroker
• SubscriptionBroker
• ConsensusPrefetchingQueue
• SubscriptionPrefetchingQueue
• SubscriptionReceiverV1