Fix write retries, Pipe resource cleanup, subscription progress, and snapshot recovery

[Caideyipi]

Description

This PR fixes partial-failure and recovery edge cases across the write path, Pipe, subscription progress, and ConfigNode snapshots. These bugs could replay data that had already been written, report unwritten rows as successful, leak generated TsFiles or reserved memory, retain stale state after recovery, replay pre-subscription rows, or skip the first pending row after a consensus subscription owner transfer.

Keep write retries and failure reporting consistent

• Track processed tablet rows across a type-inconsistency flush retry so only the remaining sequence/unsequence and time-partition fragments are retried.
• Preserve successful rows when retrying row batches, report processor-creation and insert failures at their original row indexes, and mark every uninserted tablet range with the original failure status.
• Roll back memory reserved by earlier aligned/non-aligned row groups or tablet ranges when a later reservation fails. This keeps memory accounting, last-cache updates, and consensus retry decisions consistent with the rows actually inserted.
• Keep recovered write metadata consistent by taking the maximum end time across TsFile resources and clearing per-partition memory accounting with the last-flush-time map.

Make Pipe conversion and transfer failure-safe

• Buffer table-model tablets into the V2 fallback builder and remove partial primary-path outputs before either V2 builder falls back. The tree-model builder also skips individual null schemas and accepts null DATE values without dropping valid columns or forcing fallback.
• Delete temporary TsFiles created before writer initialization, multi-database sealing, partial batch conversion, or sync/air-gap/async transfer failures.
• Complete the async error path when scheduling fails or a client has already returned to the pool, and release readers, memory blocks, event references, and generated batch files during handler or subscription cleanup. Source TsFiles remain untouched.

Preserve subscription commit progress safely

• Use versioned, encoded composite keys for consumer group, topic, and region identifiers so delimiters and numeric DataNode ID prefixes cannot collide. During rolling upgrades, write and merge unambiguous legacy keys while preferring the versioned form.
• Remove persisted progress when a consumer group unsubscribes from a topic, and advance the persistence throttle when broadcast progress is applied.
• Read snapshot fields fully: a completely absent commit-progress section remains compatible with older snapshots, while truncated headers or entries fail loading instead of silently losing progress.

Synchronize consensus subscription initial frontiers

• Register consensus queues before taking their initial frontier, roll the WAL to flush asynchronous writer metadata, and capture a stable live-WAL version/metadata pair. Header-only WAL files created by the roll are skipped instead of being reported as broken.
• Treat each local WAL-tail snapshot as a proposal. Subscription creation performs a required pull from every DataNode and publishes an explicit per-region ConfigNode result before the first poll can proceed. An explicitly empty frontier is distinguished from a missing or unavailable frontier.
• Set up new consensus topics before publishing consumer-group metadata and roll back queues and local progress if setup fails, so a partially initialized subscription cannot become visible.
• Fence preferred-writer activation until ConfigNode returns authoritative progress. Explicit absence is accepted only for a Region known to have been created after the subscription; after an explicit entry is observed, later absence is fenced again.
• When replay lookup reaches the end of sealed WAL, preserve the queue-registration boundary instead of jumping to a later live search index. Recovery writer progress still filters covered records, while pending records at the boundary remain deliverable after owner transfer.
• Return writable Thrift progress buffers and serialize ConfigNode progress snapshots consistently so explicit empty progress can cross the RPC boundary.

Replace stale state during snapshot loading

• Replace all derived template indexes and quota usage state, and clear empty template-preset or external-service snapshots instead of retaining values created after the snapshot.
• Close listening queues before deserialization so old events and references are released, reset empty linked-queue indexes and links, and rebuild ConfigNode Pipe listener references from the loaded Pipe metadata.

Verification

• mvn spotless:apply -pl iotdb-core/node-commons,iotdb-core/datanode,iotdb-core/confignode
• Targeted unit tests: CommitProgressKeeperTest, QuotaInfoTest, TemplateTableTest, TemplatePreSetTableTest, PipeTsFileBuilderV2Test, DataRegionTest, and TsFileProcessorTest.
• Additional regression coverage was added for ConfigNode commit-progress merging, Pipe listener and queue snapshot restoration, Pipe/subscription file cleanup, subscription commit state, external-service snapshots, and last-flush-time accounting.
• Consensus-frontier regression suite: 62 tests across CommitProgressKeeperTest, ConfigManagerCommitProgressTest, CommitProgressSyncProcedureTest, CreateSubscriptionProcedureTest, SubscriptionConsumerAgentTest, ConsensusPrefetchingQueueTest, ConsensusSubscriptionCommitStateTest, and ConsensusSubscriptionSetupHandlerTest.
• IoTDBConsensusSubscriptionColumnFilterClusterIT#testAlterColumnFilterRebindsAfterOwnerTransferOnThreeDataNodes passed three consecutive cluster runs.
• Full reactor Chinese-locale compile: mvn -o -nsu test-compile -DskipTests -Pwith-zh-locale.
• Integration-test reactor install, Spotless, Checkstyle, RAT, and git diff --check passed.

---

This PR has:

• been self-reviewed.
• added comments explaining the "why" and the intent where it would not be obvious.
• added unit tests or modified existing tests to cover the new code paths.

---

Key changed/added classes (or packages if there are too many classes) in this PR

• DataRegion, TsFileProcessor, and HashLastFlushTimeMap
• Pipe TsFile builders, batches, sinks, and async transfer handlers
• CommitProgressKeeper, ConsensusSubscriptionCommitManager, ConsensusSubscriptionSetupHandler, ConsensusPrefetchingQueue, and ProgressWALIterator
• ConfigManager, CommitProgressSyncProcedure, and CreateSubscriptionProcedure
• TemplateTable, TemplatePreSetTable, QuotaInfo, and ExternalServiceInfo
• PipeConfigRegionListener, AbstractSerializableListeningQueue, and ConcurrentIterableLinkedQueue