fix: correct truncate boundary of consume queues

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com>
apache · Oct 20, 2024 · 2d271ef · 2d271ef
1 parent 5cd677f
commit 2d271ef
Showing 1 changed file with 8 additions and 1 deletion.
diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java b/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java
@@ -209,6 +209,7 @@ public void forEach(Function<OffsetEntry, Boolean> predicate, Consumer<OffsetEnt
                 total = iterator.value(valueBuffer);
                 if (total != Long.BYTES + Long.BYTES) {
                     // Skip system checkpoint topic as its value is only 8 bytes
+                    iterator.next();
                     continue;
                 }
                 long commitLogOffset = valueBuffer.getLong();
@@ -394,7 +395,13 @@ public void truncateDirty(long offsetToTruncate) throws RocksDBException {
             if (entry.type == OffsetEntryType.MINIMUM) {
                 return false;
             }
-            return entry.commitLogOffset > offsetToTruncate;
+            // Normal entry offset MUST have the following inequality
+            // entry commit-log offset + message-size-in-bytes <= offsetToTruncate;
+            // otherwise, the consume queue contains dirty records to truncate;
+            //
+            // If the broker node is configured to use async-flush, it's possible consume queues contain
+            //  pointers to message records that is not flushed and lost during restart.
+            return entry.commitLogOffset >= offsetToTruncate;
         };
 
         Consumer<OffsetEntry> fn = entry -> {