[Bug] A compaction round interrupted in the scratch folder splices stale bytes into the live compaction log of the next round

Start in store/src/main/java/org/apache/rocketmq/store/kv/CompactionLog.java at compaction(...), where the scratch log is built with COMPACTING_SUB_FOLDER, and at TopicPartitionLog.clean(MappedFileQueue), which only checks the in-memory file list. Reproduce with the existing CompactionLogTest harness by pre-creating a leftover file in the compacting folder and checking whether it survives the round. Done means the scratch folder is empty before the round writes to it, the same fix covers the replicating scratch folder in pullMessageFromMaster, and the regression test passes.

Before Creating the Bug Report

• I found a bug, not just asking a question, which should be created in GitHub Discussions.
• I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.
• I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.

Runtime platform environment

OS: any (broker hosting a compaction topic; the compaction round runs on a schedule, default compactionScheduleInternal = 15 min)

RocketMQ version

branch: develop
Git commit id: 0dd48c39d

JDK Version

Compiler: Temurin 1.8.0_504 (target 1.8, same as CI)

Describe the Bug

Each compaction round writes its output into a scratch sub-folder and only moves it into the live compaction log when the round completes:

• CompactionLog.compaction(...) creates the scratch log with compacting = new TopicPartitionLog(this, COMPACTING_SUB_FOLDER) (store/src/main/java/org/apache/rocketmq/store/kv/CompactionLog.java:682); the constructor only wires the paths (:992-1006) — it neither loads nor cleans the folder;
• the scratch folder is cleaned only on the success path: replaceFiles(...) (:618) moves the files out and compacting.clean(false, false) (:621) then verifies the folder is empty and throws otherwise (TopicPartitionLog.clean(MappedFileQueue), :1095-1104). That check walks the in-memory file list, which is empty for a freshly constructed scratch log, so files left on disk are invisible to it.

So a round that is aborted after it started writing the scratch folder leaves that folder dirty and nothing ever notices:

• the round can be aborted by a broker restart or kill -9 (the round is scheduled in the background; CompactionStore.shutdown waits 1 s and then calls shutdownNow), or by any exception inside the round — compactAndReplace has no finally and doCompaction swallows the throwable.

The next round then corrupts the live compaction log, because the scratch file name is reused:

1. the new scratch MappedFileQueue starts empty, so the first append creates/opens the same file name (…/compacting/00000000000000000000) — MappedFileQueue.tryCreateMappedFile constructs a DefaultMappedFile over the existing path, which opens it rw and maps it without truncating and with wrotePosition = 0;
2. the new content overwrites the file from offset 0 and putEndMessage(compacting.getLog()) (:674-679) declares the file full;
3. whatever the aborted round had written beyond the new content is still there, and replaceFiles / replaceCqFiles (:707, :755) move that file into the live compaction log / consume queue.

The result is a compacted view that contains records from two rounds mixed together (with offsets from the aborted round), i.e. duplicated or stale messages for the compaction topic, and a consume queue whose tail does not belong to the log content.

Steps to Reproduce

1. Create a topic with the compaction attribute and let a compaction round run (or force one) until it has written into <compactionLogPath>/<topic>/<queueId>/compacting/.
2. Abort the round before it finishes the replace step (restart the broker, kill -9, or make it throw) — the scratch folder keeps its files.
3. Let the next round run: it reopens the same scratch file and appends the new compacted content on top of the old bytes, then moves it into the live log.

At the unit level the defect is proven with the existing CompactionLogTest harness (no RocksDB involved): pre-create …/<topic>/<queueId>/compacting/00000000000000000000 as a leftover, run compaction(...) with putEndMessage stubbed, and observe that the leftover file still exists afterwards — verified today with assertFalse("a scratch file left by an interrupted round must not survive into the next round", leftover.exists()), which fails.

What Did You Expect to See?

A scratch folder that is guaranteed to be empty when a round starts — the same guarantee TopicPartitionLog.clean(...) is written to check after a round. With that, an aborted round can at worst lose the messages it was compacting (they are re-compacted by the next round) instead of mixing its output into the live log.

What Did You See Instead?

The leftover files of the aborted round are silently reused and their stale bytes end up inside the live compaction log / consume queue, so consumers of the compaction topic can see duplicated or stale records.

Additional Context

Minimal fix: empty the scratch sub-folder right after the scratch log is constructed, for both scratch folders that follow this pattern (COMPACTING_SUB_FOLDER in compaction(...), REPLICATING_SUB_FOLDER in pullMessageFromMaster(...), :202), by deleting the two scratch store paths (scratch.getLog().getStorePath() and scratch.getCQ().getMappedFileQueue().getStorePath()). Clean rounds are unaffected — the folders are empty then, and the existing post-round clean(...) check keeps working.

I will open a PR with that change plus the regression test above (red before: the leftover survives the round; green after: the scratch folder is emptied before the round writes into it).