[ISSUE #10945] Align POP retry records with result batches

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

@@ -432,9 +432,10 @@ public CompletableFuture<PopConsumerContext> popAsync(String clientHost, long po
                         this.popConsumerStore.writeRecords(result.getPopConsumerRecordList());
                     }
 
+                    int popConsumerRecordIndex = 0;
                     for (int i = 0; i < result.getGetMessageResultList().size(); i++) {
                         GetMessageResult getMessageResult = result.getGetMessageResultList().get(i);
-                        PopConsumerRecord popConsumerRecord = result.getPopConsumerRecordList().get(i);
+                        PopConsumerRecord popConsumerRecord = result.getPopConsumerRecordList().get(popConsumerRecordIndex);
 
                         // If the buffer belong retries message, the message needs to be re-encoded.
                         // The buffer should not be re-encoded when popResponseReturnActualRetryTopic
@@ -445,6 +446,7 @@ public CompletableFuture<PopConsumerContext> popAsync(String clientHost, long po
                                 getMessageResult, popConsumerRecord.getTopicId(),
                                 popConsumerRecord.getQueueId(), result.getPopTime(), invisibleTime));
                         }
+                        popConsumerRecordIndex += getMessageResult.getMessageQueueOffset().size();
                     }
                 }
                 return CompletableFuture.completedFuture(result);

broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java

@@ -58,6 +58,7 @@
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 import org.apache.rocketmq.store.exception.ConsumeQueueException;
 import org.apache.rocketmq.store.stats.BrokerStatsManager;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -314,6 +315,51 @@ public void popAsyncTest() {
             20000, groupId, topicId, -1, 10, false, attemptId, ConsumeInitMode.MIN, null).join();
     }
 
+    @Test
+    public void popAsyncRecodeRetryMessagesAfterMultiMessageNormalResultTest() {
+        BrokerConfig brokerConfig = brokerController.getBrokerConfig();
+        brokerConfig.setPopResponseReturnActualRetryTopic(true);
+        brokerConfig.setPopFromRetryProbability(0);
+
+        TopicConfigManager topicConfigManager = brokerController.getTopicConfigManager();
+        SubscriptionGroupManager subscriptionGroupManager = brokerController.getSubscriptionGroupManager();
+        SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
+        Mockito.when(subscriptionGroupManager.findSubscriptionGroupConfig(groupId)).thenReturn(subscriptionGroupConfig);
+
+        String retryTopicV1 = KeyBuilder.buildPopRetryTopicV1(topicId, groupId);
+        Mockito.when(topicConfigManager.selectTopicConfig(topicId)).thenReturn(new TopicConfig(topicId, 1, 1,
+            PermName.PERM_READ | PermName.PERM_WRITE, 0));
+        Mockito.when(topicConfigManager.selectTopicConfig(retryTopicV1)).thenReturn(new TopicConfig(retryTopicV1, 1, 1,
+            PermName.PERM_READ | PermName.PERM_WRITE, 0));
+
+        GetMessageResult normalResult = getFoundResult(1L, 2L);
+        GetMessageResult retryResult = getFoundResult(3L);
+
+        PopConsumerService consumerServiceSpy = Mockito.spy(consumerService);
+        Mockito.doReturn(CompletableFuture.completedFuture(normalResult)).when(consumerServiceSpy)
+            .getMessageAsync(clientHost, groupId, topicId, 0, 0, 3, null);
+        Mockito.doReturn(CompletableFuture.completedFuture(retryResult)).when(consumerServiceSpy)
+            .getMessageAsync(clientHost, groupId, retryTopicV1, 0, 0, 1, null);
+
+        GetMessageResult recodedRetryResult = new GetMessageResult();
+        Mockito.doReturn(recodedRetryResult).when(consumerServiceSpy).recodeRetryMessage(
+            Mockito.eq(retryResult), Mockito.eq(retryTopicV1), Mockito.eq(0L), Mockito.anyLong(), Mockito.eq(20000L));
+
+        PopConsumerContext context = consumerServiceSpy.popAsync(clientHost, System.currentTimeMillis(),
+            20000, groupId, topicId, -1, 3, false, attemptId, ConsumeInitMode.MIN, null).join();
+
+        Assert.assertSame(recodedRetryResult, context.getGetMessageResultList().get(1));
+    }
+
+    private GetMessageResult getFoundResult(long... offsets) {
+        GetMessageResult result = new GetMessageResult();
+        result.setStatus(GetMessageStatus.FOUND);
+        for (long offset : offsets) {
+            result.addMessage(Mockito.mock(SelectMappedBufferResult.class), offset);
+        }
+        return result;
+    }
+
     @Test
     public void ackAsyncTest() {
         long current = System.currentTimeMillis();
@@ -743,4 +789,4 @@ public void testReviveRetryWithSuspendFalseMultipleTimes() {
             messageExt.setReconsumeTimes(capturedMessage.getReconsumeTimes());
         }
     }
-}
+}