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

broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java

@@ -22,6 +22,7 @@
 import com.google.common.cache.CacheBuilder;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.processor.PopLiteMessageProcessor;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.ServiceThread;
 import org.apache.rocketmq.common.constant.LoggerName;
@@ -45,6 +46,36 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 
+/**
+ * Routes LMQ arrival events to Lite Pop clients.
+ * maintaining a per-client bounded event queue,
+ * and waking long-polling consumers when new messages become available.
+ * useless for grpc client.
+ *
+ * <p>Key responsibilities:
+ * <ul>
+ *   <li>{@link #dispatch} selects one client from a (group, lmq) subscriber
+ *   list (with random rotation plus exclusion of the recently-removed
+ *   client) and appends the LMQ name to that client's
+ *   {@link ClientEventSet}.</li>
+ *   <li>{@link #scan} runs periodically to (a) evict event sets whose
+ *   client appears inactive or whose queue is full and (b) execute
+ *   delayed full-dispatch tasks to recover from event-queue
+ *   overflow.</li>
+ *   <li>{@link #tryDispatchToClient} / {@link ClientEventSet} implement
+ *   the per-client event queue with a dual data structure for
+ *   deduplication and a soft-cap that can be refreshed at runtime
+ *   via the {@code liteEventCapacityCacheTtlMs} TTL.</li>
+ *   <li>{@link LiteCtlListenerImpl} bridges subscription-state changes
+ *   to dispatch: registration schedules a full dispatch, removal of
+ *   all subscriptions re-dispatches the orphaned events to other
+ *   clients in the same group.</li>
+ * </ul>
+ *
+ * <p>The class is also the producer of the long-polling wake-up signal:
+ * after a successful client selection it calls
+ * {@link PopLiteMessageProcessor#getPopLiteLongPollingService PopLiteMessageProcessor.getPopLiteLongPollingService}{@code .notifyMessageArriving(...)}.
+ */
 public class LiteEventDispatcher extends ServiceThread {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LITE_LOGGER_NAME);

broker/src/main/java/org/apache/rocketmq/broker/lite/LiteShardingImpl.java

@@ -28,6 +28,17 @@
 
 import java.util.List;
 
+/**
+ * Default {@link LiteSharding} implementation: assigns each LMQ to a broker
+ * queue via consistent hashing of the lite topic segment. The sharding
+ * result is consumed by {@code AbstractLiteLifecycleManager} to decide
+ * whether a subscription should be accepted locally.
+ *
+ * <p>When the parent topic has no known route, or the LMQ name does not
+ * parse into a lite topic, the current broker is returned as a fallback so
+ * that subscription requests are not silently dropped during transient
+ * routing gaps.
+ */
 public class LiteShardingImpl implements LiteSharding {
 
     private final BrokerController brokerController;
@@ -38,6 +49,14 @@ public LiteShardingImpl(BrokerController brokerController, TopicRouteInfoManager
         this.topicRouteInfoManager = topicRouteInfoManager;
     }
 
+    /**
+     * Compute the broker that owns the given LMQ via consistent hashing of
+     * the lite topic segment over the parent's write queues.
+     * Falls back to the current broker name when
+     *  - the parent route is missing,
+     *  - or has no queues,
+     *  - or the LMQ name is not a valid lite topic.
+     */
     @Override
     public String shardingByLmqName(String parentTopic, String lmqName) {
         TopicPublishInfo topicPublishInfo = topicRouteInfoManager.tryToFindTopicPublishInfo(parentTopic);
@@ -53,6 +72,7 @@ public String shardingByLmqName(String parentTopic, String lmqName) {
         if (StringUtils.isEmpty(liteTopic)) {
             return brokerController.getBrokerConfig().getBrokerName();
         }
+
         int bucket = Hashing.consistentHash(liteTopic.hashCode(), writeQueues.size());
         MessageQueue targetQueue = writeQueues.get(bucket);
         return targetQueue.getBrokerName();

broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java

@@ -48,13 +48,40 @@
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 import org.apache.rocketmq.remoting.protocol.header.NotifyUnsubscribeLiteRequestHeader;
 
+/**
+ * Maintains the client → LMQ subscription mapping for Lite Topics.
+ *
+ * <p>Two main indexes are kept in sync:
+ * <ul>
+ *   <li>{@link #client2Subscription} — clientId → {@link LiteSubscription}
+ *   (group, topic, lmqName set) for COMPLETE-mode reconciliation</li>
+ *   <li>{@link #liteTopic2Group} — lmqName → set of {@link ClientGroup}
+ *   for fast subscription lookup during Pop dispatch</li>
+ * </ul>
+ *
+ * <p>Supports four subscription actions: PARTIAL_ADD, PARTIAL_REMOVE,
+ * COMPLETE_ADD, COMPLETE_REMOVE. When a group is configured for
+ * exclusive subscription ({@code isSubLiteExclusive}), subscribing a
+ * different client to the same lmqName evicts the previous client
+ * (tracked via {@link ExclusiveEvictionTombstones} for re-send
+ * convergence). Wildcard groups are supported by mapping them to a
+ * synthetic mock lmqName ({@code topic@group}).
+ *
+ * <p>A background thread ({@link #run}) periodically removes
+ * subscriptions whose last update exceeds the configured timeout.
+ */
 public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteSubscriptionRegistry {
     private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LITE_LOGGER_NAME);
 
+    /** Active clientId → Netty channel, used to push unsubscribe notifications. */
     protected final ConcurrentMap<String/*clientId*/, Channel> clientChannels = new ConcurrentHashMap<>();
+
+    /** Primary forward index: clientId → its full subscription (group, topic, lmq set). */
     protected final ConcurrentMap<String/*clientId*/, LiteSubscription> client2Subscription = new ConcurrentHashMap<>();
     protected final ConcurrentMap<String/*lmqName*/, Set<ClientGroup>> liteTopic2ClientGroup = new ConcurrentHashMap<>();
     protected final ConcurrentMap<String/*topic*/, Set<String/*group*/>> wildcardGroupMap = new ConcurrentHashMap<>();
+
+    /** Cached expansion of wildcard group clients per group, 30s TTL. */
     private final Cache<String/*group*/, List<ClientGroup>> wildcardClientCache =
         CacheBuilder.newBuilder().maximumSize(2000).expireAfterWrite(30, TimeUnit.SECONDS).build();
 
@@ -79,14 +106,40 @@ public void updateClientChannel(String clientId, Channel channel) {
         clientChannels.put(clientId, channel);
     }
 
+    /**
+     * Add a partial subscription for {@code lmqNameSet} to the given client,
+     * union with whatever the client already subscribes to.
+     *
+     * <p>Rejects the call with {@link LiteQuotaException} if the total active
+     * (client, liteTopic) reference count would exceed
+     * {@code maxLiteSubscriptionCount}, and with {@link IllegalStateException}
+     * if the group is a wildcard group (use {@link #addCompleteSubscription}
+     * instead).
+     *
+     * <p>For each lmqName that is still subscription-active, the call:
+     * <ol>
+     *   <li>Adds the lmqName to the client's {@link LiteSubscription};</li>
+     *   <li>In exclusive mode, evicts the previous holder of the same
+     *   (group, lmqName) pair and clears any stale tombstone for the
+     *   caller;</li>
+     *   <li>Applies {@code offsetOption} via {@link #resetOffset};</li>
+     *   <li>Registers the (clientId, group) in
+     *   {@link #liteTopic2Group}.</li>
+     * </ol>
+     */
     @Override
     public void addPartialSubscription(String clientId, String group, String topic, Set<String> lmqNameSet,
         OffsetOption offsetOption) {
+        // default maxLiteSubscriptionCount is 100000
         long maxCount = brokerController.getBrokerConfig().getMaxLiteSubscriptionCount();
         if (getActiveSubscriptionNum() >= maxCount) {
             // No need to check existence, if reach here, it must be new.
             throw new LiteQuotaException("lite subscription quota exceeded " + maxCount);
         }
+
+        // Wildcard group is not supported
+        // Wildcard group receives all LMQ message from a parent topic
+        // Wildcard group is same as normal group
         if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
             throw new IllegalStateException("subscribe lite operation is not supported for this group");
         }
@@ -97,6 +150,7 @@ public void addPartialSubscription(String clientId, String group, String topic,
         final ClientGroup clientGroup = new ClientGroup(clientId, thisSub.getGroup());
 
         for (String lmqName : lmqNameSet) {
+            // subscription exists or sharding to this broker
             if (!liteLifecycleManager.isSubscriptionActive(topic, lmqName)) {
                 continue;
             }
@@ -108,6 +162,8 @@ public void addPartialSubscription(String clientId, String group, String topic,
                 // subsequent popLiteTopic is not blocked by the stale mark.
                 exclusiveEvictionTombstones.remove(clientId, lmqName);
             }
+
+            // set subscription start offset
             resetOffset(lmqName, group, clientId, offsetOption);
             addClientGroup(thisSub, clientGroup, lmqName);
         }
@@ -124,6 +180,20 @@ public void removePartialSubscription(String clientId, String group, String topi
         }
     }
 
+    /**
+     * Replace the client's full subscription set with {@code lmqNameAll}
+     * (filtered to active lmqNames). For wildcard groups, the new set is
+     * reduced to a single synthetic lmqName and the wildcard marker is
+     * registered.
+     *
+     * <p>The implementation diffs the previous set against the new one:
+     * lmqNames no longer in the new set are removed from the forward and
+     * reverse indexes; lmqNames added are registered. In exclusive groups,
+     * any leftover eviction tombstones for the client are either re-notified
+     * (if the lmqName is still in the new set, indicating the previous
+     * unsubscribe notification was lost) or cleaned up (if the lmqName is no
+     * longer subscribed).
+     */
     @Override
     public void addCompleteSubscription(String clientId, String group, String topic, Set<String> lmqNameAll, long version) {
         if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
@@ -325,7 +395,11 @@ protected void removeClientGroup(ClientGroup clientGroup, String lmqName, boolea
     }
 
     /**
-     * Remove clients that subscribe to the same liteTopic under the same group
+     * Remove clients that subscribe to the same liteTopic under the same group.
+     * - remove liteTopic from client2Subscription by clientId
+     * - add exclusive tombstones
+     * - call client to notify unsubscribe event
+     * - call removeTopicGroup
      */
     protected void excludeClientByLmqName(String newClientId, String group, String lmqName) {
         Set<ClientGroup> clientSet = liteTopic2ClientGroup.get(lmqName);
@@ -345,8 +419,10 @@ protected void excludeClientByLmqName(String newClientId, String group, String l
                     client2Subscription.remove(clientGroup.clientId);
                 }
             }
+
             exclusiveEvictionTombstones.add(clientGroup.clientId, lmqName);
             notifyUnsubscribeLite(clientGroup.clientId, clientGroup.group, lmqName);
+
             boolean resetOffset = LiteMetadataUtil.isResetOffsetInExclusiveMode(group, brokerController);
             LOGGER.info("excludeClientByLmqName group:{}, lmqName:{}, resetOffset:{}, clientId:{} -> {}",
                 group, lmqName, resetOffset, clientGroup.clientId, newClientId);
@@ -511,4 +587,4 @@ public boolean hasExclusiveEvictionTombstone(String clientId, String lmqName) {
         return exclusiveEvictionTombstones.contains(clientId, lmqName);
     }
 
-}
+}

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,15 @@ public class PopConsumerCache extends ServiceThread {
     private final Consumer<PopConsumerRecord> reviveConsumer;
 
     private final AtomicInteger estimateCacheSize;
+    /**
+     * Maps to the buffered records for that consumer-queue.
+     * - key: {@code consumerGroupId@topicId@queueId}
+     * - value: {@code ConsumerRecords}
+     *
+     * <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 +133,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()));
@@ -127,8 +182,7 @@ public int cleanupRecords(Consumer<PopConsumerRecord> consumer) {
 
             if (timeout) {
                 records.stageExpiredRecords(Long.MAX_VALUE);
-                List<PopConsumerRecord> writeConsumerRecords =
-                    new ArrayList<>(records.getRemoveTreeMap().values());
+                List<PopConsumerRecord> writeConsumerRecords = new ArrayList<>(records.getRemoveTreeMap().values());
                 if (!writeConsumerRecords.isEmpty()) {
                     consumerRecordStore.writeRecords(writeConsumerRecords);
                 }
@@ -205,13 +259,45 @@ 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) {