add comments to pop/ack/changInvisibleTime, No Code change

broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLongPollingService.java

@@ -52,6 +52,24 @@
 import static org.apache.rocketmq.broker.longpolling.PollingResult.POLLING_SUC;
 import static org.apache.rocketmq.broker.longpolling.PollingResult.POLLING_TIMEOUT;
 
+/**
+ * Pop-mode long polling service that suspends Pop requests and wakes them up when new messages arrive.
+ * <p>
+ * Core responsibilities:
+ * <ul>
+ *   <li>Suspend Pop requests — when the broker has no messages to return immediately, registers requests
+ *       into the {@code pollingMap} keyed by {@code topic@cid@queueId} and waits</li>
+ *   <li>Wake up on new message arrival — {@link #notifyMessageArriving} is triggered by the message arriving
+ *       listener; it fetches matching Pop requests from the pollingMap, applies Tag filtering, and re-submits
+ *       them to the PopMessageProcessor to return results to the client</li>
+ *   <li>Timeout scanning — the background thread periodically scans the waiting queues and wakes up
+ *       timed-out requests with an empty result</li>
+ *   <li>Retry topic bridging — {@link #notifyMessageArrivingFromRetry} translates a new message on the retry
+ *       topic into a wake-up notification on the original topic</li>
+ *   <li>Resource cleanup — periodically removes stale polling entries for deleted topics or
+ *       offline consumer groups</li>
+ * </ul>
+ */
 public class PopLongPollingService extends ServiceThread {
 
     private static final Logger POP_LOGGER =

broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java

@@ -35,6 +35,41 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * In-memory cache for un-acked Pop consumer records, used when
+ * {@code enablePopBufferMerge} is enabled in the KVStore path.
+ *
+ * <p>
+ * The cache structure is as follows: {
+ *     groupId@topicId@queueId: {
+ *         active: ConcurrentSkipListMap<offset, PopConsumerRecord>,
+ *         removed: ConcurrentSkipListMap<offset, PopConsumerRecord>
+ *     }
+ * }
+ * active(recordTreeMap): in-flight records
+ * removed(removedTreeMap): records to be removed
+ * </p>
+ *
+ * <p>Popped messages are stored here by
+ * {@link PopConsumerService#popAsync}. The background {@link #run()} thread
+ * periodically scans the cache and processes expired records:
+ * <ul>
+ *   <li><b>Visibility timeout expired</b> — the record is passed to the
+ *       {@code reviveConsumer} (which calls
+ *       {@link PopConsumerService#revive}) to re-publish the message to
+ *       the retry topic</li>
+ *   <li><b>Consumer offline</b> (lock timeout) — all records for that
+ *       {code groupId, topicId} pair are flushed to
+ *       {@link PopConsumerKVStore} without revival</li>
+ *   <li><b>Consumer acked</b> — the record is removed via
+ *       {@link #deleteRecords} when a matching ack arrives</li>
+ * </ul>
+ *
+ * <p>Each {@code groupId@topicId@queueId} entry is backed by a
+ * {@link ConsumerRecords} instance containing two
+ * {@link ConcurrentSkipListMap}s — one for active records and one for
+ * records staged for removal.
+ */
 public class PopConsumerCache extends ServiceThread {
 
     private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LOGGER_NAME);
@@ -46,6 +81,14 @@ public class PopConsumerCache extends ServiceThread {
     private final Consumer<PopConsumerRecord> reviveConsumer;
 
     private final AtomicInteger estimateCacheSize;
+    /**
+     * Maps {@code consumerGroupId@topicId@queueId} to the buffered records for that
+     * consumer-queue.
+     *
+     * <p>Used by {@link #writeRecords} to add popped messages,
+     * {@link #deleteRecords} to remove acked messages, and
+     * {@link #cleanupRecords} to process expired records.
+     */
     private final ConcurrentMap<String, ConsumerRecords> consumerRecordTable;
 
     public PopConsumerCache(BrokerController brokerController, PopConsumerKVStore consumerRecordStore,
@@ -89,9 +132,20 @@ public long getPopInFlightMessageCount(String groupId, String topicId, int queue
         return consumerRecords != null ? consumerRecords.getInFlightRecordCount() : 0L;
     }
 
+    /**
+     * Write popped records into the cache.
+     *
+     * <p>Each record is inserted into the {@link ConsumerRecords} for its
+     * {@code groupId@topicId@queueId}. If no entry exists for that key, a
+     * new one is created. The cache size estimate is incremented.
+     *
+     * @param consumerRecordList the popped records to cache
+     */
     public void writeRecords(List<PopConsumerRecord> consumerRecordList) {
         this.estimateCacheSize.addAndGet(consumerRecordList.size());
         consumerRecordList.forEach(consumerRecord -> {
+            // consumerRecords is the recordMap in cache
+            // it contains two maps of PopConsumerRecord
             ConsumerRecords consumerRecords = ConcurrentHashMapUtils.computeIfAbsent(consumerRecordTable,
                 this.getKey(consumerRecord), k -> new ConsumerRecords(brokerController.getBrokerConfig(),
                     consumerRecord.getGroupId(), consumerRecord.getTopicId(), consumerRecord.getQueueId()));
@@ -205,13 +259,46 @@ public void run() {
         }
     }
 
+    /**
+     * Records for one {@code consumerGroupId@topicId@queueId} in the Pop cache.
+     *
+     * <p>Uses two {@link ConcurrentSkipListMap}s to separate active and
+     * expiring records for safe two-phase cleanup:
+     * <ol>
+     *   <li>{@link #stageExpiredRecords} moves timed-out records from
+     *       {@link #recordTreeMap} to {@link #removeTreeMap}</li>
+     *   <li>{@link PopConsumerCache#cleanupRecords} drains
+     *       {@link #removeTreeMap} — true-expired records are revived,
+     *       approaching-expired records are written to the KVStore</li>
+     * </ol>
+     */
     protected static class ConsumerRecords {
 
         private final String groupId;
         private final String topicId;
         private final int queueId;
         private final BrokerConfig brokerConfig;
+        /**
+         * Staged records awaiting cleanup (revival or KVStore write).
+         *
+         * <p>Populated by {@link #stageExpiredRecords} and drained by
+         * {@link PopConsumerCache#cleanupRecords}. Sorted by offset
+         * so that {@link #getMinOffset} can include these records in
+         * the minimum offset computation.
+         */
         private final ConcurrentSkipListMap<Long /* offset */, PopConsumerRecord> removeTreeMap;
+        /**
+         * Active (in-flight) records that have been popped but not yet
+         * acked by the consumer.
+         *
+         * <p>Records are added via {@link #write} when messages are popped,
+         * removed via {@link #delete} when an ack arrives, and moved to
+         * {@link #removeTreeMap} via {@link #stageExpiredRecords} when
+         * the visibility timeout or stay-buffer time expires.
+         *
+         * <p>Sorted by offset for efficient minimum-offset queries
+         * ({@link #getMinOffset}).
+         */
         private final ConcurrentSkipListMap<Long /* offset */, PopConsumerRecord> recordTreeMap;
 
         public ConsumerRecords(BrokerConfig brokerConfig, String groupId, String topicId, int queueId) {

broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerKVStore.java

@@ -18,6 +18,22 @@
 
 import java.util.List;
 
+/**
+ * Persistent key-value store for un-acked Pop consumer records.
+ *
+ * <p>Used by the KVStore-based ack path ({@code popConsumerKVServiceEnable=true}).
+ * When a message is popped, a record is written here. When the consumer acks
+ * the message or the visibility timeout expires, the record is deleted or
+ * revived. The default implementation is {@code PopConsumerRocksdbStore}.
+ *
+ * <p>This interface supports three operations:
+ * <ul>
+ *   <li>{@link #writeRecords} — persist popped records</li>
+ *   <li>{@link #deleteRecords} — remove acked records</li>
+ *   <li>{@link #scanExpiredRecords} — find records whose visibility timeout
+ *       has elapsed for revival</li>
+ * </ul>
+ */
 public interface PopConsumerKVStore {
 
     /**

broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerRecord.java

@@ -58,6 +58,15 @@ public int getCode() {
     @JSONField(ordinal = 4)
     private int retryFlag;
 
+    /**
+     * Message visibility timeout in milliseconds.
+     *
+     * <p>The visibility timeout ({@code popTime + invisibleTime}) determines when
+     * a popped-but-unacked message becomes eligible for revival. Set by the
+     * consumer (default 60s via {@code DefaultMQPushConsumer#setPopInvisibleTime}).
+     * Can be changed by proxy with config.
+     * Can be extended via {@code ChangeInvisibleTime}.
+     */
     @JSONField(ordinal = 5)
     private long invisibleTime;
 
@@ -67,9 +76,31 @@ public int getCode() {
     @JSONField(ordinal = 7)
     private int attemptTimes;
 
+    /**
+     * Client-generated idempotency key for FIFO ordered consumption.
+     *
+     * <p>Possible values:
+     * <ul>
+     *   <li>Client request — a unique id from {@code PopMessageRequestHeader#getAttemptId},
+     *       used by {@code ConsumerOrderInfoManager} to block subsequent pops on the same
+     *       queue until the current batch is acked</li>
+     *   <li>{@code null} — for ack records ({@code PopConsumerService#ackAsync}) and
+     *       change-invisibility records, where FIFO ordering is not applicable</li>
+     *   <li>Copied from the original record — for revive retry records, the attemptId is
+     *       inherited from the expired record</li>
+     * </ul>
+     */
     @JSONField(ordinal = 8)
     private String attemptId;
 
+    /**
+     * Whether the consumer has suspended (nacked) this message.
+     *
+     * <p>When {@code true}, the reconsume count is <b>not</b> incremented on
+     * revive, so the message will not be prematurely sent to the DLQ due to
+     * repeated visibility timeout extensions. Set via
+     * {@code ChangeInvisibleTimeRequestHeader#isSuspend}.
+     */
     @JSONField(ordinal = 9)
     private boolean suspend;
 
@@ -102,7 +133,19 @@ public long getVisibilityTimeout() {
     }
 
     /**
-     * Key: timestamp(8) + groupId + topicId + queueId + offset
+     * Build the RocksDB key for this record.
+     *
+     * <p>Format:
+     * <pre>
+     * visibilityTimeout(8B) + groupId + '@' + topicId + '@' + queueId(4B) + '@' + offset(8B)
+     * </pre>
+     *
+     * <p>The {@code visibilityTimeout} is placed first so that records are ordered
+     * by expiration time in RocksDB's SST files. This allows
+     * {@code PopConsumerRocksdbStore#scanExpiredRecords} to use a bounded iterator
+     * to scan only the relevant time window without a full table scan.
+     *
+     * <p>NACK(changeInvisibleTime) will create a new record, and the old one will be deleted.
      */
     @JSONField(serialize = false)
     public byte[] getKeyBytes() {

broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerRocksdbStore.java

@@ -38,6 +38,19 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * RocksDB-backed implementation of {@link PopConsumerKVStore} for the
+ * KVStore-based Pop ack path.
+ *
+ * <p>Stores Pop consumer records in a dedicated {@code "popState"} column
+ * family. Each record is keyed by {@code visibilityTimeout|groupId@topicId@queueId@offset}
+ * so that {@link #scanExpiredRecords} can efficiently scan only expired
+ * records within a time window without a full table scan.
+ *
+ * <p>Write and delete operations use synchronous flush and WAL for
+ * durability — Pop visibility state is the sole source of truth in the
+ * KVStore path and must survive crashes.
+ */
 public class PopConsumerRocksdbStore extends AbstractRocksDBStorage implements PopConsumerKVStore {
 
     private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LOGGER_NAME);
@@ -55,31 +68,71 @@ public PopConsumerRocksdbStore(String filePath, long blockCacheSize, long writeB
         this.writeBufferSize = writeBufferSize;
     }
 
-    // https://www.cnblogs.com/renjc/p/rocksdb-class-db.html
-    // https://github.com/johnzeng/rocksdb-doc-cn/blob/master/doc/RocksDB-Tuning-Guide.md
+    /**
+     * Configure RocksDB options for Pop consumer record storage.
+     *
+     * <p>Unlike the parent class defaults, write and delete options enable
+     * WAL and synchronous flush — Pop visibility state is the sole source
+     * of truth and must survive crashes. Compaction is configured to be
+     * aggressive so that expired-then-deleted records are purged promptly,
+     * reclaiming disk space.
+     *
+     * @see <a href="https://www.cnblogs.com/renjc/p/rocksdb-class-db.html">rocksdb-class-db</a>
+     * @see <a href="https://github.com/johnzeng/rocksdb-doc-cn/blob/master/doc/RocksDB-Tuning-Guide.md">RocksDB-Tuning-Guide</a>
+     */
     protected void initOptions() {
+        // durability-first: enable WAL and sync flush for pop state recovery
         this.options = RocksDBOptionsFactory.createDBOptions();
 
         this.writeOptions = new WriteOptions();
+        // fsync every write to disk
         this.writeOptions.setSync(true);
+        // enable WAL
         this.writeOptions.setDisableWAL(false);
+        // allow writing throttling under pressure
         this.writeOptions.setNoSlowdown(false);
 
+        // delete must be durable too — otherwise ack can be lost and message revived incorrectly
         this.deleteOptions = new WriteOptions();
         this.deleteOptions.setSync(true);
         this.deleteOptions.setDisableWAL(false);
         this.deleteOptions.setNoSlowdown(false);
 
+        // aggressive compaction to purge expired pop records and reclaim space
         this.compactRangeOptions = new CompactRangeOptions();
+        // force compact bottom level
         this.compactRangeOptions.setBottommostLevelCompaction(
             CompactRangeOptions.BottommostLevelCompaction.kForce);
+        // allow compaction to pause writes
         this.compactRangeOptions.setAllowWriteStall(true);
+        // manual compaction runs in parallel with auto-compaction.
+        // Appropriate here because expired Pop records generate tombstones continuously,
+        // and cleanup should not starve RocksDB's normal background work
         this.compactRangeOptions.setExclusiveManualCompaction(false);
+        // Allows compaction to move data across levels
         this.compactRangeOptions.setChangeLevel(true);
+        // -1 delegates level selection to RocksDB's internal heuristics
         this.compactRangeOptions.setTargetLevel(-1);
+        // Splits the compaction work into at most 4 parallel sub-tasks
         this.compactRangeOptions.setMaxSubcompactions(4);
     }
 
+    /**
+     * Initialise the RocksDB instance with a dedicated column family for Pop state.
+     *
+     * <p>Two column families are created:
+     * <ol>
+     *   <li>{@code default} — unused, required by RocksDB</li>
+     *   <li>{@code "popState"} — stores Pop consumer records keyed by
+     *       {@code visibilityTimeout|groupId@topicId@queueId@offset}</li>
+     * </ol>
+     *
+     * <p>Called by {@link AbstractRocksDBStorage#start()} before the storage
+     * is marked as loaded. Returns {@code false} if any step fails, preventing
+     * all subsequent read/write operations via {@link #hold()}.
+     *
+     * @return {@code true} if the database was opened successfully
+     */
     @Override
     protected boolean postLoad() {
         try {
@@ -111,6 +164,16 @@ public String getFilePath() {
         return this.dbPath;
     }
 
+    /**
+     * Batch-write consumer records to RocksDB via a single {@link WriteBatch}.
+     * Key: (popTime + invisibleTime) + groupId + topicId + queueId + offset
+     * value: PopConsumerRecord.toJsonBytes
+     *
+     * <p>Each record is serialized with its visibility-timeout-prefixed key
+     * so that {@link #scanExpiredRecords} can efficiently scan by time range.
+     *
+     * @param consumerRecordList the records to persist
+     */
     @Override
     public void writeRecords(List<PopConsumerRecord> consumerRecordList) {
         if (!consumerRecordList.isEmpty()) {
@@ -125,6 +188,14 @@ public void writeRecords(List<PopConsumerRecord> consumerRecordList) {
         }
     }
 
+    /**
+     * Batch-delete consumer records from RocksDB via a single {@link WriteBatch}.
+     *
+     * <p>Deletion uses the same durability guarantees as writes ({@code sync=true},
+     * WAL enabled)
+     *
+     * @param consumerRecordList the records to remove
+     */
     @Override
     public void deleteRecords(List<PopConsumerRecord> consumerRecordList) {
         if (!consumerRecordList.isEmpty()) {
@@ -139,8 +210,19 @@ public void deleteRecords(List<PopConsumerRecord> consumerRecordList) {
         }
     }
 
+    /**
+     * Scan and return expired consumer records within a visibility-timeout range.
+     *
+     * <p>Because each record's key is prefixed with {@code visibilityTimeout},
+     * this method uses a RocksDB iterator bounded by {@code [lower, upper)} to
+     * efficiently scan only the relevant time window without a full table scan.
+     *
+     * @param lower    inclusive lower bound of the visibility timeout (ms)
+     * @param upper    exclusive upper bound of the visibility timeout (ms)
+     * @param maxCount maximum number of records to return
+     * @return up to {@code maxCount} expired records, or an empty list
+     */
     @Override
-    // https://github.com/facebook/rocksdb/issues/10300
     public List<PopConsumerRecord> scanExpiredRecords(long lower, long upper, int maxCount) {
         // In RocksDB, we can use SstPartitionerFixedPrefixFactory in cfOptions
         // and new ColumnFamilyOptions().useFixedLengthPrefixExtractor() to
@@ -153,6 +235,7 @@ public List<PopConsumerRecord> scanExpiredRecords(long lower, long upper, int ma
              RocksIterator iterator = db.newIterator(this.columnFamilyHandle, scanOptions)) {
             iterator.seek(ByteBuffer.allocate(Long.BYTES).putLong(lower).array());
             while (iterator.isValid() && consumerRecordList.size() < maxCount) {
+                // decode json bytes to PopConsumerRecord
                 consumerRecordList.add(PopConsumerRecord.decode(iterator.value()));
                 iterator.next();
             }