[Subscription] Reduce consensus WAL replay and ACK contention

[Caideyipi]

Description

This PR removes repeated DataNode-side WAL replay work and reduces the ACK lock contention amplified by that work. It only changes IoTDB/DataNode code; no consumer code is changed.

Bottleneck shown by the flame graphs

The four CPU/wall profiles show one server-side root cause and its client-side amplification.

On the DataNode CPU profile:

• Consensus subscription prefetch accounts for about 79.3% of CPU.
• The WAL input path under DataInputStream.readFully accounts for about 60.4%.
• LZ4_decompress_safe self time is about 32.3%.
• openReaderAtIndex is about 33.5% cumulative and skipEntries is about 16.2%.
• Actual low-level disk reads are only a few percent.

The bottleneck is therefore repeated WAL replay work, rather than raw disk throughput. tryCatchUpFromWAL() previously reset the iterator before every batch. A preceding hasNext() could already have parsed and cached nextReady, but the next round closed that iterator, discarded the cached request, reopened the WAL at nextExpectedSearchIndex, and repeated reader lookup, skipping, reads, and LZ4 decompression.

Even after keeping the iterator alive, every round still listed, parsed, filtered, and sorted all retained WAL files. Live-WAL reopen also copied the same metadata twice: once to check whether new entries existed and again while opening the reader.

On the DataNode wall profile, the expensive prefetch round holds the queue read lock. A late/missing ACK then waits for the queue write lock for roughly 86% of its RPC wall time.

On the consumer wall profile, the auto-commit RPC waits for that ACK response while holding the consumer SynchronizedHandler monitor. A concurrent business thread calling commitSync waits for the same monitor for roughly 48% of consumer wall time. Enabling auto-commit together with manual commitSync amplifies the server stall, but the server-side WAL replay and ACK lock contention are the root causes addressed here.

The consumer CPU profile also shows Tablet.deserialize at about 49% and Tablet.readValuesFromBuffer at about 40%. Those are separate consumer-side costs and are intentionally outside this DataNode-only PR.

Implemented low-risk optimizations

• Keep ProgressWALIterator and its hasNext() cache alive across normal WAL catch-up rounds instead of reopening it at the current search index for every batch.
• Refresh the retained WAL file list only after the current iterator is exhausted. Do not perform a second redundant refresh immediately after rolling the WAL and creating a fresh iterator.
• Cache the sorted WAL version IDs alongside the file array. Refresh relocation and version lookup now use binary search instead of repeatedly parsing file names and linearly scanning the array.
• Reuse the already-fetched live-WAL WALMetaData snapshot when reopening the reader, avoiding a second full metadata copy for the same reopen attempt.
• Request an explicit deferred iterator realignment when the realtime pending path advances the local cursor independently. Seek, gap recovery, memory retry, close, and lifecycle transitions retain their explicit reset behavior.
• While a reset is pending, scheduling reports immediate work without calling hasNext() on the stale iterator. The next prefetch round applies the reset under the queue lock.
• Handle late/missing ACKs under the queue read lock. The touched queue/map indexes are concurrent containers, event cleanup is idempotently protected by the event monitor, and commit-state updates are serialized by the commit-state monitor. The read lock still fences seek/close and other lifecycle transitions that require the queue write lock.

Performance test

ProgressWALIteratorTest#testIteratorReusePerformance is disabled by default and can be enabled manually:

mvn test -pl iotdb-core/datanode \
  -Dtest=ProgressWALIteratorTest#testIteratorReusePerformance \
  -Diotdb.test.subscription.performance=true

Optional parameters:

• -Diotdb.test.subscription.performance.entries=<count>
• -Diotdb.test.subscription.performance.batch-size=<count>

With 4,096 entries and batches of 64 on the same generated WAL:

reopen=928.971 ms
refreshEachBatch=32.641 ms
reuse=19.792 ms
reopenSpeedup=46.94x
refreshSpeedup=1.65x

reopen models the original per-batch reopen/skip/decompress behavior. refreshEachBatch keeps the iterator but still rescans retained WAL files on every batch. reuse is the final path in this PR. The benchmark isolates these WAL iterator costs from consumer RPC and tablet deserialization.

High-risk directions intentionally not implemented

• Metadata-first filtering that skips entry bodies: this crosses WAL format/metadata compatibility and subscription progress semantics, including body search-index fallback and fragmented-request grouping.
• Special-casing skipToEntryIndex(0): bypassing the current call changes when the first WAL segment is read and validated, which can alter corrupted-WAL and near-live retry behavior.
• An incremental live-WAL reader: dynamically extending reader end offsets must remain correct across EOF, rollover, compression, encryption, and partially visible bytes.
• A shared decompressed-segment cache: this requires explicit direct-memory accounting, reference counting, eviction, and coordination with WAL deletion/rollover lifecycles.
• A ByteBuffer pool/slab for consensus requests: IoTConsensusRequest can retain buffers beyond the reader call, so safe reuse requires a new ownership/lifetime contract.
• Moving WAL I/O and tablet conversion outside the queue lock: this needs concurrency redesign around seek generations, memory reservation, publish failures, and rollback so stale batches cannot escape.
• Adaptive pending retention/backpressure: per-subscription memory bounds and fairness across multiple consumers need a policy-level design and are not a deterministic local hot-path change.

These directions may provide additional gains, but they carry correctness or memory-lifecycle risk disproportionate to this focused optimization and should be handled separately with dedicated stress/compatibility testing.

Tests

• ConsensusPrefetchingQueueTest#testLateAckDoesNotWaitForPrefetchReadLock
• ConsensusPrefetchingQueueTest#testWalCatchUpReusesIteratorAcrossRounds
• ConsensusPrefetchingQueueTest#testReadableWalIteratorSkipsFileListRefresh
• ConsensusPrefetchingQueueTest#testWalRollDoesNotRefreshNewIteratorTwice
• ConsensusPrefetchingQueueTest#testPendingCursorAdvanceDefersWalIteratorRealignment
• ProgressWALIteratorTest#testLiveWalReopenReusesMetadataSnapshot
• Full ConsensusPrefetchingQueueTest: 25 passed
• Full ProgressWALIteratorTest: 9 discovered, 0 failures, 0 errors, 1 performance test skipped by default
• Manual performance test: 1 passed
• Spotless and Checkstyle passed

---

This PR has:

• been self-reviewed.
  • concurrent read
  • concurrent write
  • concurrent read and write
• added comments explaining the why and the intent of the code wherever it would not be obvious for an unfamiliar reader.
• added unit tests to cover the new code paths.

---

Key changed/added classes

• ConsensusPrefetchingQueue
• ProgressWALIterator
• ConsensusPrefetchingQueueTest
• ProgressWALIteratorTest

[Caideyipi]

Could we clear or update this deferred reset when fillGapFromWAL realigns the iterator? For example, with pending requests [1, 5], processing 1 records a reset target of 2. Processing 5 then fills the WAL gap from 2 and may advance both the iterator and nextExpectedSearchIndex to 6. Since request 5 is now before the local cursor, this method is not called again, so the pending target remains 2. The next prefetch round applies that stale target and rereads/skips/decompresses WAL entries 2 through 5, recreating the replay cost this PR is intended to remove. More generally, applying this reset at the start of every following round also reconstructs ProgressWALIterator (including listing and sorting retained WAL files) once per steady-state pending batch even when that round never needs WAL. Please consider consuming the marker only immediately before entering the WAL path, and clearing/updating it when fillGapFromWAL has already aligned the iterator. A regression test with pending [1, 5] and WAL [2..5] would cover the stale-target case.