[Subscription] Balance consensus ownership and report WAL backlog

[Caideyipi]

Description

Balance and retain consensus subscription ownership

Consensus subscription previously selected each Region owner with
floorMod(regionId.hashCode(), activeConsumerCount). Hash collisions can leave consumers without
any Region even when the consumer count does not exceed the Region count. For example, the
reproduced 40-Region/40-consumer topology assigned no Region to 9 consumers. When those consumers
stopped polling, each membership change also rehashed every Region and caused cascading ownership
churn.

This change builds a balanced, sticky ownership snapshot:

• Preserve assignments whose Region and consumer are both still active.
• Assign new Regions and Regions from departed consumers to the least-loaded active consumer.
• Move existing assignments only until consumer loads differ by at most one.
• Use deterministic consumer and Region ordering for tie-breaking.
• Serialize ownership refresh with queue lifecycle changes and retain the snapshot when only one
  Region is removed.

As a result, every consumer owns at least one Region whenever the consumer count is no greater than
the Region count, while joins, leaves, and Region changes move only the assignments required to
restore balance.

Report the unread WAL backlog in lag

The queue lag previously collapsed any unread WAL backlog to one extra unit. A queue with hundreds
of thousands of unread WAL search indexes could therefore report lag=1.

The updated lag adds already materialized lifecycle stages separately and uses the larger of the
pending-entry count and raw WAL search-index gap for the unmaterialized stage. Pending local entries
overlap the raw WAL gap until they advance the cursor, so this avoids double-counting while
preserving the actual backlog magnitude.

This PR deliberately does not change the global prefetch executor size or shared subscription
memory limits; those are separate scheduling and admission-policy decisions.

Verification

mvn -pl iotdb-core/datanode -Dtest=ConsensusSubscriptionBrokerOwnershipTest,ConsensusPrefetchingQueueTest#testLagIncludesLingeringBatchUntilCommitted+testLagIncludesUnreadWalSearchIndexDistance -DfailIfNoTests=false test

Tests run: 8, failures: 0, errors: 0, skipped: 0.

---

This PR has:

• been self-reviewed.
  • concurrent read
  • concurrent write
  • concurrent read and write
• added comments explaining the why and intent where it is not obvious.
• added or updated unit tests for the changed code paths.

---

Key changed/added classes

• ConsensusSubscriptionBroker
• ConsensusPrefetchingQueue
• ConsensusSubscriptionBrokerOwnershipTest
• ConsensusPrefetchingQueueTest