[Caideyipi]

Summary

• Add poll-side backpressure for pipe-based subscription queues when too many events are already in flight. Before rejecting a poll, the queue first remaps in-flight events so committed or pollable entries can be cleaned/recycled and immediately free capacity.
• Enforce subscription prefetch limits against retained events instead of only prefetched events. Retained events include both queued and in-flight events, so slow or stalled consumers cannot bypass the local/global event caps by continuously moving events out of the prefetch queue.
• Apply the same retained-event capacity accounting to consensus subscription queues across polling, WAL/realtime prefetch, lingered batch emission, watermark emission, and idle rescheduling.
• Wake consensus prefetch after ack/ackSilent frees a retained-event slot, so queues resume promptly after backpressure clears.
• Inject the prefetching queue count into SubscriptionPrefetchingQueueStates for focused unit coverage of global threshold calculations without bootstrapping the global subscription broker.

Motivation

Previously, the main limits primarily looked at prefetched queue size. If a consumer kept polling but did not commit, events moved into the in-flight map and no longer counted against those prefetch limits. That could allow retained subscription events to accumulate and delay backpressure until memory pressure or other secondary limits were hit. This PR makes in-flight events part of the capacity model and throttles further polling when the in-flight side itself reaches the configured limits.

Testing

• mvn spotless:apply -pl iotdb-core/datanode
• git diff --check origin/master...HEAD
• mvn -Ddevelocity.off=true -pl iotdb-core/datanode -Dtest=SubscriptionPrefetchingQueueStatesTest -DforkCount=0 test
  • Ran 4 tests in SubscriptionPrefetchingQueueStatesTest.
  • forkCount=0 was used only for this local Windows environment because the default Surefire fork JVM failed to start due to insufficient page file / native memory before any tests ran.

[Caideyipi]

Self-review passed after fixes in 9589f42. All actionable findings from this review have been resolved.

Fixed findings:

• Concurrent polls could pass the in-flight limit check together and then exceed the limit. The limit check and event claim now share a dedicated lock in both pipe and consensus queues. Deterministic regression tests reproduce the race for pipe V1 and V2.
• Consensus prefetch checked retained capacity at round entry, but inner pending/WAL, buffered-writer, and ordered-writer paths could still over-materialize or exceed the event cap. Those paths now account for in-flight events and reserve a slot for the lingering batch, while preserving the existing byte-based memory behavior.
• Full queues scheduled recycling using the global interval instead of the topic override, including when topic recycling was disabled. Scheduling now honors the effective topic interval and becomes dormant for nonpositive intervals.
• Event-capacity backpressure incorrectly disabled the separately bounded realtime admission queue, forcing unnecessary WAL fallback. Materialization now pauses without disabling that admission path.

Also removed the two newly introduced untranslated debug message literals. Added regression coverage for pending/WAL/multi-writer delivery, ack/ackSilent recovery, retained capacity, and topic recycle overrides.

Validation:

• Reproduced 12 failures among 18 selected cases against the original PR head, 0733a17.
• After the fixes: 120 tests passed across all 10 SubscriptionPrefetchingQueue*Test and ConsensusPrefetchingQueue*Test classes (0 failures, 0 ignored).
• Compiled the affected production classes and the selected test classes with javac --release 17; also compiled the affected production classes with the Chinese message sources. Tests ran through JUnit 4 against the installed Maven dependency classpath, with the JVM opens required by the existing reflection-based test fixtures.
• mvn spotless:apply -P with-integration-tests -pl iotdb-core/datanode -Ddevelocity.off=true and git diff --check passed.
• Full Maven reactor and integration tests were not run. The remote Maven attempt could not resolve its build extension because Maven Central was unreachable and the internal mirror DNS lookup failed; verification used the focused local compilation/JUnit fallback described above.