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/PopConsumerService.java

@@ -298,6 +298,31 @@ public boolean isFifoBlocked(PopConsumerContext context, String groupId, String
             context.getAttemptId(), topicId, groupId, queueId, context.getInvisibleTime());
     }
 
+    /**
+     * Fetch messages from a single queue and append them to the pop context.
+     *
+     * <p>Chained via {@link CompletableFuture#thenCompose} from
+     * {@link #getMessageFromTopicAsync}. When the batch is already full
+     * ({@code remain <= 0}), the pending count is added to the context and
+     * the chain stops. Otherwise, messages are fetched from the store and
+     * the result is merged into the context via {@link #handleGetMessageResult}.
+     *
+     * <p>Early termination can occur inside this method when:
+     * <ul>
+     *   <li>Too many inflight (un-acked) messages exist</li>
+     *   <li>A FIFO queue is blocked</li>
+     * </ul>
+     *
+     * @param future    the accumulator future carrying the pop context
+     * @param clientHost the client address
+     * @param groupId   consumer group id
+     * @param topicId   topic name
+     * @param queueId   queue id
+     * @param batchSize max number of messages still needed
+     * @param filter    message filter
+     * @param retryType whether this is a retry topic V1/V2
+     * @return a future completing with the pop context updated with results
+     */
     protected CompletableFuture<PopConsumerContext> getMessageAsync(CompletableFuture<PopConsumerContext> future,
         String clientHost, String groupId, String topicId, int queueId, int batchSize, MessageFilter filter,
         PopConsumerRecord.RetryType retryType) {
@@ -335,13 +360,38 @@ protected CompletableFuture<PopConsumerContext> getMessageAsync(CompletableFutur
         });
     }
 
+    /**
+     * Fetch messages from every read queue of a topic via a CompletableFuture chain.
+     *
+     * <p>Each queue is visited once. For each queue the
+     * {@link #getMessageAsync(CompletableFuture, String, String, String, int, int, MessageFilter, PopConsumerRecord.RetryType)}
+     * method is chained via {@link CompletableFuture#thenCompose}. The chain carries
+     * the accumulated result through all queues, stopping early when the batch is
+     * filled, the queue is blocked, or the inflight threshold is reached.
+     *
+     * <p>Queue iteration order respects {@code priorityOrderAsc} and uses
+     * {@code requestCount} as a round-robin offset for load balancing.
+     *
+     * @param future       the accumulator future
+     * @param clientHost   the client address
+     * @param groupId      consumer group id
+     * @param topicId      topic name
+     * @param requestCount round-robin counter for queue selection
+     * @param batchSize    max number of messages to return
+     * @param filter       message filter expression
+     * @param retryType    whether this is a retry topic V1/V2
+     * @return a future completing with the pop result context
+     */
     protected CompletableFuture<PopConsumerContext> getMessageFromTopicAsync(CompletableFuture<PopConsumerContext> future,
         String clientHost, String groupId, String topicId, long requestCount, int batchSize, MessageFilter filter,
         PopConsumerRecord.RetryType retryType) {
+        // get topic config
         TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(topicId);
         if (null == topicConfig) {
             return future;
         }
+
+        // iterate all queues of the topic
         for (int i = 0; i < topicConfig.getReadQueueNums(); i++) {
             long index = (brokerController.getBrokerConfig().isPriorityOrderAsc() ?
                 topicConfig.getReadQueueNums() - 1 - i : i) + requestCount;
@@ -352,10 +402,38 @@ protected CompletableFuture<PopConsumerContext> getMessageFromTopicAsync(Complet
         return future;
     }
 
+    /**
+     * Asynchronously pop messages for the KVStore-based ack path.
+     *
+     * <p>This method coordinates the full Pop lifecycle:
+     * <ol>
+     *   <li>Validates topic, group, and acquires the consumer lock</li>
+     *   <li>Determines whether to pull from retry topic first
+     *       (based on {@code popFromRetryProbability})</li>
+     *   <li>Pulls messages from normal topic (and retry topic V1/V2 if configured)</li>
+     *   <li>Writes checkpoints to {@link PopConsumerCache} (buffer merge) or
+     *       {@link PopConsumerKVStore} (RocksDB)</li>
+     *   <li>Re-encodes retry messages if needed</li>
+     * </ol>
+     *
+     * @param clientHost   the client address
+     * @param popTime      the pop invocation timestamp
+     * @param invisibleTime the message visibility timeout
+     * @param groupId      consumer group id
+     * @param topicId      topic name
+     * @param queueId      queue id (-1 for all queues)
+     * @param batchSize    max number of messages to return
+     * @param fifo         whether this is a FIFO ordered consumption
+     * @param attemptId    attempt id for idempotent consumption
+     * @param initMode     consume init mode (min/max)
+     * @param filter       message filter expression
+     * @return a future that completes with the pop result context
+     */
     public CompletableFuture<PopConsumerContext> popAsync(String clientHost, long popTime, long invisibleTime,
         String groupId, String topicId, int queueId, int batchSize, boolean fifo, String attemptId, int initMode,
         MessageFilter filter) {
 
+        // init context params
         PopConsumerContext popConsumerContext =
             new PopConsumerContext(clientHost, popTime, invisibleTime, groupId, fifo, initMode, attemptId);
 
@@ -391,25 +469,30 @@ public CompletableFuture<PopConsumerContext> popAsync(String clientHost, long po
             CompletableFuture.completedFuture(popConsumerContext);
 
         try {
+            // get message from retry topic,
             if (!fifo && preferRetry) {
+                // default config of retrieveMessageFromPopRetryTopicV1 is true,
                 if (brokerConfig.isRetrieveMessageFromPopRetryTopicV1()) {
                     getMessageFuture = this.getMessageFromTopicAsync(getMessageFuture, clientHost, groupId,
                         retryTopicV1, requestCount, batchSize, filter, PopConsumerRecord.RetryType.RETRY_TOPIC_V1);
                 }
 
+                // default config of enableRetryTopicV2 is false
                 if (brokerConfig.isEnableRetryTopicV2()) {
                     getMessageFuture = this.getMessageFromTopicAsync(getMessageFuture, clientHost, groupId,
                         retryTopicV2, requestCount, batchSize, filter, PopConsumerRecord.RetryType.RETRY_TOPIC_V2);
                 }
             }
 
+            // get message from normal topic
             if (queueId != -1) {
                 getMessageFuture = this.getMessageAsync(getMessageFuture, clientHost, groupId,
                     topicId, queueId, batchSize, filter, PopConsumerRecord.RetryType.NORMAL_TOPIC);
             } else {
                 getMessageFuture = this.getMessageFromTopicAsync(getMessageFuture, clientHost, groupId,
                     topicId, requestCount, batchSize, filter, PopConsumerRecord.RetryType.NORMAL_TOPIC);
 
+                // get message from retry topic
                 if (!fifo && !preferRetry) {
                     if (brokerConfig.isRetrieveMessageFromPopRetryTopicV1()) {
                         getMessageFuture = this.getMessageFromTopicAsync(getMessageFuture, clientHost, groupId,
@@ -425,13 +508,16 @@ public CompletableFuture<PopConsumerContext> popAsync(String clientHost, long po
 
             return getMessageFuture.thenCompose(result -> {
                 if (result.isFound() && !result.isFifo()) {
+                    // write checkpoint to cache or store
+                    // default config of enablePopBufferMerge is false
                     if (brokerConfig.isEnablePopBufferMerge() &&
                         popConsumerCache != null && !popConsumerCache.isCacheFull()) {
                         this.popConsumerCache.writeRecords(result.getPopConsumerRecordList());
                     } else {
                         this.popConsumerStore.writeRecords(result.getPopConsumerRecordList());
                     }
 
+                    // format result
                     for (int i = 0; i < result.getGetMessageResultList().size(); i++) {
                         GetMessageResult getMessageResult = result.getGetMessageResultList().get(i);
                         PopConsumerRecord popConsumerRecord = result.getPopConsumerRecordList().get(i);
@@ -449,6 +535,7 @@ public CompletableFuture<PopConsumerContext> popAsync(String clientHost, long po
                 }
                 return CompletableFuture.completedFuture(result);
             }).whenComplete((result, throwable) -> {
+                // unlock by consumerLockService
                 try {
                     if (throwable != null) {
                         log.error("PopConsumerService popAsync get message error",