diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java index f72e2ba26f2..f5ca492ee73 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java @@ -432,9 +432,10 @@ public CompletableFuture 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 popAsync(String clientHost, long po getMessageResult, popConsumerRecord.getTopicId(), popConsumerRecord.getQueueId(), result.getPopTime(), invisibleTime)); } + popConsumerRecordIndex += getMessageResult.getMessageQueueOffset().size(); } } return CompletableFuture.completedFuture(result); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java index 44189744b46..b3b78eb0622 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java +++ b/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()); } } -} \ No newline at end of file +}