leizhiyuan opened a new issue, #11495: URL: https://github.com/apache/rocketmq/issues/11495
### Before Creating the Bug Report - [X] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq/discussions). - [X] I have searched the [GitHub Issues](https://github.com/apache/rocketmq/issues) and [GitHub Discussions](https://github.com/apache/rocketmq/discussions) of this repository and believe that this is not a duplicate. - [X] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ. ### Runtime platform environment Reproduced on macOS (arm64) and Linux. Not platform specific. ### RocketMQ version branch: develop Git commit id: 592d59d12f (latest develop at the time of writing) Broker configured with `storeType=defaultRocksDB` (RocksDB ConsumeQueue). rocksdbjni 8.4.4 as declared in `pom.xml`. ### JDK Version OpenJDK 21 (also reproducible on JDK 8; the issue is independent of the JDK). ### Describe the Bug With the RocksDB ConsumeQueue store, the min offset of a newly created queue can be computed and **persisted** as a value greater than 0 (for example 3 or 4) while the queue is being written. After that, `getMessage` returns `OFFSET_TOO_SMALL` for the real first offsets, consumers correct their offset to the wrong min offset, and the first messages of the queue are silently skipped. Nothing is logged at error level and the consumer lag goes back to 0, so the lost messages are hard to notice. **Root cause** `RocksDBOptionsFactory#createOffsetCFOptions()` enables `setInplaceUpdateSupport(true)` on the `offset` column family. `RocksDBConsumeQueueStore#putMessagePosition0` writes, in **one** `WriteBatch`, the new CQ units to the default CF and the queue's MAX offset key to the offset CF. The code assumes this batch becomes visible atomically. When in-place update is enabled and the MAX key already exists in the memtable, RocksDB overwrites the value **in place**. That overwrite is not protected by sequence-number visibility. A concurrent reader can therefore see the **new** MAX before the CQ units of the same batch are visible, because the CQ units are normal inserts that only become visible once the batch publishes its sequence number. As a result, the batch is no longer atomic for readers. The first min lookup of a queue (no MIN key yet) then does the following in `RocksDBConsumeQueueOffsetTable#correctMinCqOffset`: 1. `getMaxPhyAndCqOffsetInKV` returns the already overwritten MAX = `m`. 2. `RocksDBConsumeQueueTable#binarySearchInCQ(high = m, low = 0, min = true)`: entries of the in-flight batch return `null`, and the loop moves forward with `low = mid + 1`. It ends with either `-1` or a cq offset that is too large. 3. On `-1`, `updateCqOffset(topic, queueId, maxPhyOffset, maxCQOffset, false)` persists MIN = `m` (and a too-large result is persisted the same way). The wrong MIN is written to RocksDB and cached in `topicQueueMinOffset`. It is never recomputed, because `isMinOffsetOk` now considers it valid. **Impact** - Any broker using `storeType=defaultRocksDB`. - Most likely on newly created queues, especially LMQ queues, which are created on first write. It is more likely when many consumers long-poll: a message arrival wakes many requests that compute the min offset while the next dispatch batch is being written. - Silent message loss, with no error log and lag reported as 0. - Other readers that read MAX from the offset CF and then read CQ units are exposed to the same torn read, e.g. `correctMaxCqOffset` and `getOffsetInQueueByTime`. - In a test environment we saw new queues get min offsets of 4, 13 and 10 right after their first writes, and consumers skipped exactly those messages. ### Steps to Reproduce Standalone program, plain rocksdbjni 8.4.4 (the version used by develop). Its options mirror the broker: `disableWAL`, `manualWalFlush`, `allowConcurrentMemtableWrite=false`, `atomicFlush`. A writer puts `CQ[i]` (default CF) and `MAX=i` (offset CF) in one batch. Readers read MAX and then point-get `CQ[MAX]`. ```java import java.nio.ByteBuffer; import java.nio.file.Files; import java.util.*; import java.util.concurrent.atomic.*; import org.rocksdb.*; public class InplaceVisibilityRepro { public static void main(String[] args) throws Exception { boolean inplace = Boolean.parseBoolean(args[0]); long seconds = Long.parseLong(args[1]); RocksDB.loadLibrary(); ColumnFamilyOptions cqOpt = new ColumnFamilyOptions(); ColumnFamilyOptions offOpt = new ColumnFamilyOptions().setInplaceUpdateSupport(inplace); DBOptions dbOpt = new DBOptions().setCreateIfMissing(true).setCreateMissingColumnFamilies(true) .setManualWalFlush(true).setAllowConcurrentMemtableWrite(false).setAtomicFlush(true); List<ColumnFamilyHandle> h = new ArrayList<>(); RocksDB db = RocksDB.open(dbOpt, Files.createTempDirectory("inplace").toString(), Arrays.asList( new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cqOpt), new ColumnFamilyDescriptor("offset".getBytes(), offOpt)), h); ColumnFamilyHandle cq = h.get(0), off = h.get(1); WriteOptions wo = new WriteOptions().setSync(false).setDisableWAL(true); ReadOptions ro = new ReadOptions(); byte[] maxKey = "max".getBytes(); AtomicBoolean stop = new AtomicBoolean(); AtomicLong reads = new AtomicLong(), anomalies = new AtomicLong(); List<Thread> threads = new ArrayList<>(); for (int t = 0; t < 4; t++) { threads.add(new Thread(() -> { try { while (!stop.get()) { byte[] max = db.get(off, ro, maxKey); // read MAX if (max == null) continue; reads.incrementAndGet(); if (db.get(cq, ro, max) == null) { // CQ[MAX] of the same batch anomalies.incrementAndGet(); } } } catch (RocksDBException e) { throw new RuntimeException(e); } })); } threads.forEach(Thread::start); long deadline = System.currentTimeMillis() + seconds * 1000; for (long i = 0; System.currentTimeMillis() < deadline; i++) { byte[] k = ByteBuffer.allocate(8).putLong(i).array(); try (WriteBatch wb = new WriteBatch()) { // like putMessagePosition0 wb.put(cq, k, k); wb.put(off, maxKey, k); db.write(wo, wb); } } stop.set(true); for (Thread t : threads) t.join(); System.out.println("inplace=" + inplace + " reads=" + reads + " MAX-visible-but-CQ[MAX]-missing=" + anomalies); db.close(); } } ``` ``` J=~/.m2/repository/org/rocksdb/rocksdbjni/8.4.4/rocksdbjni-8.4.4.jar javac -cp $J InplaceVisibilityRepro.java java -cp $J:. InplaceVisibilityRepro false 10 java -cp $J:. InplaceVisibilityRepro true 10 ``` Results with rocksdbjni 8.4.4 (10 seconds each): | inplaceUpdateSupport | reads | MAX visible but CQ[MAX] missing | |---|---|---| | false | 30,804,889 | **0** | | true | 4,023,367 | **18,392** | A second reproducer replays the exact `correctMinCqOffset` + `binarySearchInCQ` logic. Each round creates a fresh queue and writes batch1 = `CQ[0] + MAX(0)`, then batch2 = `CQ[1..4] + MAX(4)`, while 8 readers do the first min lookup: | inplaceUpdateSupport | rounds | rounds with computed min > 0 | min values | |---|---|---|---| | false | 300,000 | **0** | - | | true | 300,000 | **1,714** | 3 (1,085), 4 (629) | The same happens with the real store. Dispatching batch1 = `{offset 0}` and batch2 = `{offsets 1..4}` for hundreds of fresh queues through `RocksDBConsumeQueueStore#putMessagePosition`, while other threads call `getMinOffsetInQueue`, occasionally leaves `getMinOffsetInQueue(topic, 0) == 4` afterwards. ### What Did You Expect to See? The CQ units and the MAX offset written in one `WriteBatch` become visible atomically, and the min offset of a new queue is 0. ### What Did You See Instead? Readers see the new MAX before the CQ units of the same batch. The first min lookup persists a min offset > 0 (e.g. 3 or 4), and consumers skip the first messages of the queue. ### Additional Context Proposed fix: remove `setInplaceUpdateSupport(true)` from `createOffsetCFOptions()`. The offset CF only holds a few keys per queue (MAX/MIN), so normal versioned writes are cheap. Old versions are reclaimed by memtable flush and compaction, and point lookups still see only the latest visible version. I will submit a PR with a regression test. Possible follow-ups (not part of the fix): - `binarySearchInCQ` in min mode silently skips `null` entries that are `<= max`, which can hide an inconsistency instead of reporting it. - Consumers commit an `OFFSET_TOO_SMALL` correction unconditionally, so a wrong min offset directly becomes message loss. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
