[Subscription] Protect retained WAL by committed progress - #18399
[Subscription] Protect retained WAL by committed progress#18399Caideyipi wants to merge 1 commit into
Conversation
| final WALNode walNode = (WALNode) consensusReqReader; | ||
| final File[] walFiles = WALFileUtils.listAllWALFiles(walNode.getLogDirectory()); | ||
| if (Objects.isNull(walFiles) || walFiles.length == 0) { | ||
| return Math.max(0L, currentWalVersion); | ||
| } | ||
|
|
||
| WALFileUtils.ascSortByVersionId(walFiles); | ||
| for (final File walFile : walFiles) { | ||
| final long versionId = WALFileUtils.parseVersionId(walFile.getName()); | ||
| if (versionId >= currentWalVersion) { | ||
| return Math.max(0L, currentWalVersion); | ||
| } | ||
| if (ProgressWALIterator.isHeaderOnlyWalFile(walFile)) { | ||
| continue; | ||
| } | ||
|
|
||
| WalFileCommitRequirement requirement = walFileCommitRequirements.get(versionId); |
There was a problem hiding this comment.
Is it possible to record the last visited WalFileCommitRequirement, and avoid listing files when it cannot still be covered by the committed progress.
The overhead of computing the id after each commit concerns me.
There was a problem hiding this comment.
The metadata itself is already cached in walFileCommitRequirements, and refreshCommittedWalRetentionBound short-circuits when both committed progress and the current WAL version are unchanged. I kept the directory listing when either changes because a roll can introduce a new boundary and WAL cleanup can remove the previously visited file; reusing a single file id without checking that state could retain or release the wrong prefix.
| private static int compareProgress( | ||
| final WriterProgress leftProgress, final WriterProgress rightProgress) { | ||
| final int physicalTimeComparison = | ||
| Long.compare(leftProgress.getPhysicalTime(), rightProgress.getPhysicalTime()); | ||
| return physicalTimeComparison != 0 | ||
| ? physicalTimeComparison | ||
| : Long.compare(leftProgress.getLocalSeq(), rightProgress.getLocalSeq()); | ||
| } | ||
| } |
There was a problem hiding this comment.
WriterProgress does not implement compareTo?
There was a problem hiding this comment.
WriterProgress is a payload value class and does not expose compareTo. The queue intentionally keeps compareProgress local so the ordering (physical time, then local sequence) is explicit and is shared by all retention checks. No behavior change is needed here.
2 participants
Description
Retain WAL required by uncommitted consumers
Previously, subscription WAL reclamation only considered replication progress and Topic retention policies. Consumer-group committed progress was not part of the deletion boundary, so retention could reclaim WAL still required by a lagging consumer.
This PR derives a file-level retained WAL version for each
(consumerGroup, topic, region)queue from its per-writer committedRegionProgress. IoTConsensus combines these boundaries with Topic size/time retention and uses the most conservative boundary.Multi-writer and leader migration handling
Committed progress remains represented by
(physicalTime, writerNodeId, localSeq)rather than localsearchIndex. Rolled WAL V3 metadata is summarized per writer, allowing follower WAL records and records produced before leader migration to be protected correctly.WAL files with incomplete or unsupported writer metadata are conservatively retained.
Boundary refresh and caching
The committed WAL boundary is refreshed after ACK, direct commit, initialization, seek, committed-progress broadcast, or WAL roll. Per-file writer-progress requirements are cached so repeated safe-delete checks do not repeatedly scan unchanged WAL metadata.
This PR has:
Tests
mvn spotless:apply -pl iotdb-core/consensus,iotdb-core/datanodemvn test -pl iotdb-core/consensus -Dtest=SubscriptionWalRetentionCalculatorTestmvn test -pl iotdb-core/consensus,iotdb-core/datanode -Dtest=ConsensusPrefetchingQueueTest#testWalFileCommitRequirementUsesPerWriterMaximum+testWalFileCommitRequirementRejectsUnsupportedWriterMetadata -Dsurefire.failIfNoSpecifiedTests=falsemvn test-compile -pl iotdb-core/consensus,iotdb-core/datanode -P with-zh-locale -DskipTestsKey changed/added classes
SubscriptionWalRetentionCalculatorSubscriptionQueueRegistryIoTConsensusServerImplConsensusPrefetchingQueue