[luoluoyuyu]

Description

Problem

PipeTsFileEpochProgressIndexKeeper keeps pipe tsFile dedup state per data region (regionId -> taskScope -> tsFilePath -> TsFileResource). Entries are only removed by pipe event callbacks and by the realtime source's close(). StorageEngine.deleteDataRegion() removes the region, its files, WAL, snapshots and the receiver-side resource, but never removes this source-side dedup state. After a region is deleted (DROP DATABASE, region migration cleanup), its entries stay in the DataNode heap for the rest of the process lifetime: memory grows on every create/drop cycle, and a re-created region can be matched against stale progress indexes.

Solution

• Add PipeTsFileEpochProgressIndexKeeper.clearProgressIndex(int dataRegionId), which drops all task scopes of one region at once.
• Call it from StorageEngine.deleteDataRegion() right after region.markDeleted(), i.e. before the region files are removed, so the cleanup also happens when the rest of the deletion fails. The cleanup is logged through a new StorageEngineMessages constant (en and zh).

Tests

• UT PipeTsFileEpochProgressIndexKeeperTest#testClearProgressIndexRemovesAllTaskScopesOfDataRegion: entries of two different task scopes of the same region are all removed.
• UT DataRegionTest#testDeletedRegionClearsItsPipeProgressIndexes: registers an index, runs the real StorageEngine.deleteDataRegion(), asserts the entry is gone.
• IT IoTDBPipeProgressIndexCleanupIT (LocalStandaloneIT, 1C1D, SimpleIT): creates a database, writes and flushes data, drops the database, and waits for the DataNode log to report the cleanup. It fails on the pre-fix build (timeout, no cleanup record) and passes with the fix (StorageEngine:887 - Cleared pipe progress indexes of the deleted data region 1).

Verified locally: keeper UT 5/5, DataRegionTest method 1/1, IT 1/1, spotless:check for iotdb-core/datanode and integration-test, test-compile for the default locale and for -P with-zh-locale.

---

This PR has:

• been self-reviewed.
• added unit tests or modified existing tests to cover new code paths.
• added integration tests.
• been tested in a test IoTDB cluster.

---

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

• org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper
• org.apache.iotdb.db.storageengine.StorageEngine
• org.apache.iotdb.db.i18n.StorageEngineMessages (en / zh)
• org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeperTest
• org.apache.iotdb.db.storageengine.dataregion.DataRegionTest
• org.apache.iotdb.db.it.IoTDBPipeProgressIndexCleanupIT

[Caideyipi]

Reviewed the full diff and the relevant region-deletion, asynchronous Pipe assignment, keeper registration, and locale-selection paths at dec46bf. Two actionable issues are noted inline. The current GitHub checks show 37 successful checks and 2 skipped checks; I did not rerun Maven builds or tests during this review.

[Caideyipi]

[P2] Prevent pending Pipe events from repopulating the deleted region

markDeleted() stops storage-side work, but it neither deregisters the realtime Pipe sources nor drains their assigner. An insert already published to PipeDataRegionAssigner can therefore reach PipeRealtimeDataRegionHybridSource.extractTabletInsertion() after this clear and call registerProgressIndex() for the deleted region; that method recreates the region entry with computeIfAbsent. In particular, a downgraded tablet from a working TsFile can reintroduce the resource after deletion, while forceCloseAllWorkingTsFileProcessors() discards that processor without publishing its normal TsFile-close event. The entry then still depends on the later source close() for removal, so the new deletion cleanup does not guarantee that the deleted region's state stays cleared. Please coordinate the final clear with outstanding registrations, or reject registrations from the obsolete region lifecycle, and cover an event paused before registration while deleteDataRegion() completes. The added UTs only register before deletion, and the IT creates no Pipe, so none exercises this ordering.

[Caideyipi]

[P2] Make the cleanup assertion work with the Chinese locale

With -P with-zh-locale, the DataNode emits the Chinese StorageEngineMessages.LOG_CLEARED_PIPE_PROGRESS_INDEXES_OF_THE_DELETED_DATA_REGION_ARG_DC6E6D9F value added in this PR. This hardcoded English substring can never match that log, so running this IT with both with-integration-tests and with-zh-locale always times out after 60 seconds even when cleanup succeeds. Derive the expected message from the selected StorageEngineMessages constant (accounting for the region-id placeholder), or use an assertion independent of the log language. Compiling both locales alone does not exercise this failure.