RockteMQ-AI commented on code in PR #4316: URL: https://github.com/apache/rocketmq-dashboard/pull/4316#discussion_r4015108207
########## src/main/java/org/apache/rocketmq/dashboard/service/impl/DlqAutoReplayServiceImpl.java: ########## @@ -0,0 +1,126 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.dashboard.service.impl; + +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.protocol.body.GroupList; +import org.apache.rocketmq.dashboard.model.DlqAutoReplayReport; +import org.apache.rocketmq.dashboard.service.DlqAutoReplayService; +import org.apache.rocketmq.tools.admin.MQAdminExt; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +import javax.annotation.Resource; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +@Service +public class DlqAutoReplayServiceImpl implements DlqAutoReplayService { + + private static final Logger log = LoggerFactory.getLogger(DlqAutoReplayServiceImpl.class); + + @Resource + private MQAdminExt mqAdminExt; + + private final Map<String, DlqAutoReplayReport> reportCache = new ConcurrentHashMap<>(); + + @Override + public DlqAutoReplayReport getReplayStatus(String consumerGroup) { + if (StringUtils.isBlank(consumerGroup)) { + consumerGroup = "DEFAULT_GROUP"; + } + final String finalGroup = consumerGroup; + return reportCache.computeIfAbsent(consumerGroup, k -> createInitialReport(finalGroup)); + } + + @Override + public DlqAutoReplayReport executeAutoReplay(String consumerGroup, DlqAutoReplayReport.ReplayPolicy policy) { + if (StringUtils.isBlank(consumerGroup)) { + consumerGroup = "DEFAULT_GROUP"; + } + + DlqAutoReplayReport report = reportCache.computeIfAbsent(consumerGroup, this::createInitialReport); + if (policy != null) { + report.setPolicy(policy); + } else { + policy = report.getPolicy(); + } + + String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + consumerGroup; + report.setDlqTopic(dlqTopic); + + int batchCount = Math.min(policy.getMaxBatchSize(), 200); + int successCount = (int) (batchCount * 0.98); + int failCount = batchCount - successCount; + + String batchId = "BATCH-" + System.currentTimeMillis(); + long now = System.currentTimeMillis(); + + DlqAutoReplayReport.ReplayExecutionItem item = new DlqAutoReplayReport.ReplayExecutionItem( + batchId, now - 1500, now, batchCount, successCount, failCount, "admin"); + + report.getExecutionHistory().add(0, item); + report.setReplayedMessages(report.getReplayedMessages() + successCount); + report.setFailedMessages(report.getFailedMessages() + failCount); + + long totalProcessed = report.getReplayedMessages() + report.getFailedMessages(); + double successRate = totalProcessed > 0 ? ((double) report.getReplayedMessages() / totalProcessed) * 100.0 : 100.0; + report.setReplaySuccessRate(Math.round(successRate * 100.0) / 100.0); + + long remaining = Math.max(0, report.getTotalDlqMessages() - successCount); + report.setTotalDlqMessages(remaining); + report.setStatus(remaining == 0 ? "IDLE" : "REPLAYING"); + + report.getAuditLogs().add(0, String.format("[%tF %<tT] Executed auto-replay batch %s: %d succeeded, %d failed.", + now, batchId, successCount, failCount)); + + return report; + } + + private DlqAutoReplayReport createInitialReport(String consumerGroup) { + DlqAutoReplayReport report = new DlqAutoReplayReport(); Review Comment: **[Warning]** Thread safety: `executeAutoReplay` mutates the `DlqAutoReplayReport` object (incrementing `replayedMessages`, `failedMessages`, updating `lastReplayTime`, `status`, etc.) without synchronization. If two concurrent requests target the same `consumerGroup`, the counters can be corrupted. Consider synchronizing on the report object or using `AtomicLong` for counters. ########## src/main/java/org/apache/rocketmq/dashboard/service/impl/DlqAutoReplayServiceImpl.java: ########## @@ -0,0 +1,126 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.dashboard.service.impl; + +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.protocol.body.GroupList; +import org.apache.rocketmq.dashboard.model.DlqAutoReplayReport; +import org.apache.rocketmq.dashboard.service.DlqAutoReplayService; +import org.apache.rocketmq.tools.admin.MQAdminExt; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +import javax.annotation.Resource; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +@Service +public class DlqAutoReplayServiceImpl implements DlqAutoReplayService { Review Comment: **[Critical]** `reportCache` is an unbounded `ConcurrentHashMap` that grows with every unique `consumerGroup` queried. There is no eviction policy, TTL, or size limit. In a long-running dashboard instance this will leak memory. Consider using a bounded cache (e.g. Caffeine/Guava `CacheBuilder.maximumSize()` with `expireAfterWrite`) or at minimum a `LinkedHashMap` with `removeEldestEntry`. ########## src/test/java/org/apache/rocketmq/dashboard/service/impl/DlqAutoReplayServiceImplTest.java: ########## @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.dashboard.service.impl; + +import org.apache.rocketmq.common.protocol.body.GroupList; +import org.apache.rocketmq.dashboard.model.DlqAutoReplayReport; +import org.apache.rocketmq.tools.admin.MQAdminExt; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import java.util.HashSet; + +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.when; + +public class DlqAutoReplayServiceImplTest { + + @Mock + private MQAdminExt mqAdminExt; + + @InjectMocks + private DlqAutoReplayServiceImpl dlqAutoReplayService; + + @Before + public void setUp() { + MockitoAnnotations.initMocks(this); + } + + @Test + public void testGetReplayStatus() throws Exception { + GroupList groupList = new GroupList(); + groupList.setGroupList(new HashSet<>()); + when(mqAdminExt.queryTopicConsumeByWho(anyString())).thenReturn(groupList); + + DlqAutoReplayReport report = dlqAutoReplayService.getReplayStatus("test-group"); + Assert.assertNotNull(report); Review Comment: **[Info]** The test asserts `assertThat(report.getReplayedMessages()).isEqualTo(100)` which only validates the hardcoded mock logic (`batchCount * 0.98 = 100 * 0.98 ≈ 98 → cast to int`). This test does not verify any real DLQ replay behavior. When the service is updated to perform actual replay, these tests will need to be completely rewritten with mocked `mqAdminExt` interactions. ########## src/main/java/org/apache/rocketmq/dashboard/controller/DlqAutoReplayController.java: ########## @@ -0,0 +1,52 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.dashboard.controller; + +import org.apache.rocketmq.dashboard.model.DlqAutoReplayReport; +import org.apache.rocketmq.dashboard.permisssion.Permission; +import org.apache.rocketmq.dashboard.service.DlqAutoReplayService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.ResponseBody; + +@Controller +@RequestMapping("/dlq") +@Permission +public class DlqAutoReplayController { + + @Autowired + private DlqAutoReplayService dlqAutoReplayService; + + @RequestMapping(value = "/autoReplay/status.query", method = RequestMethod.GET) + @ResponseBody + public DlqAutoReplayReport getReplayStatus(@RequestParam("consumerGroup") String consumerGroup) { + return dlqAutoReplayService.getReplayStatus(consumerGroup); + } Review Comment: **[Warning]** The `consumerGroup` parameter defaults to `"DEFAULT_GROUP"` when blank. This is risky — a missing or empty parameter silently targets the default consumer group, which could trigger unintended replay operations. Consider returning a 400 error for missing `consumerGroup` instead of silently defaulting. ########## src/main/java/org/apache/rocketmq/dashboard/service/impl/DlqAutoReplayServiceImpl.java: ########## @@ -0,0 +1,126 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.dashboard.service.impl; + +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.protocol.body.GroupList; +import org.apache.rocketmq.dashboard.model.DlqAutoReplayReport; +import org.apache.rocketmq.dashboard.service.DlqAutoReplayService; +import org.apache.rocketmq.tools.admin.MQAdminExt; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +import javax.annotation.Resource; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +@Service +public class DlqAutoReplayServiceImpl implements DlqAutoReplayService { + + private static final Logger log = LoggerFactory.getLogger(DlqAutoReplayServiceImpl.class); + + @Resource + private MQAdminExt mqAdminExt; + + private final Map<String, DlqAutoReplayReport> reportCache = new ConcurrentHashMap<>(); + + @Override + public DlqAutoReplayReport getReplayStatus(String consumerGroup) { + if (StringUtils.isBlank(consumerGroup)) { + consumerGroup = "DEFAULT_GROUP"; + } + final String finalGroup = consumerGroup; + return reportCache.computeIfAbsent(consumerGroup, k -> createInitialReport(finalGroup)); + } + + @Override + public DlqAutoReplayReport executeAutoReplay(String consumerGroup, DlqAutoReplayReport.ReplayPolicy policy) { + if (StringUtils.isBlank(consumerGroup)) { + consumerGroup = "DEFAULT_GROUP"; + } + + DlqAutoReplayReport report = reportCache.computeIfAbsent(consumerGroup, this::createInitialReport); + if (policy != null) { + report.setPolicy(policy); + } else { + policy = report.getPolicy(); + } + + String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + consumerGroup; + report.setDlqTopic(dlqTopic); + + int batchCount = Math.min(policy.getMaxBatchSize(), 200); + int successCount = (int) (batchCount * 0.98); + int failCount = batchCount - successCount; + + String batchId = "BATCH-" + System.currentTimeMillis(); + long now = System.currentTimeMillis(); + + DlqAutoReplayReport.ReplayExecutionItem item = new DlqAutoReplayReport.ReplayExecutionItem( + batchId, now - 1500, now, batchCount, successCount, failCount, "admin"); + + report.getExecutionHistory().add(0, item); + report.setReplayedMessages(report.getReplayedMessages() + successCount); + report.setFailedMessages(report.getFailedMessages() + failCount); + + long totalProcessed = report.getReplayedMessages() + report.getFailedMessages(); + double successRate = totalProcessed > 0 ? ((double) report.getReplayedMessages() / totalProcessed) * 100.0 : 100.0; + report.setReplaySuccessRate(Math.round(successRate * 100.0) / 100.0); + + long remaining = Math.max(0, report.getTotalDlqMessages() - successCount); + report.setTotalDlqMessages(remaining); Review Comment: **[Critical]** `createInitialReport` hardcodes `totalDlqMessages = 150` and `executeAutoReplay` hardcodes `successCount = (int)(batchCount * 0.98)`. This is not a real DLQ replay — no messages are actually fetched from the DLQ topic or sent back to the original topic. The `mqAdminExt` is injected but never used for any replay operation. This implementation gives users a false sense that their dead-letter messages are being replayed. The service needs to: 1. Query actual DLQ message count via `mqAdminExt.queryConsumeStats` or similar 2. Actually consume messages from `%DLQ%{consumerGroup}` and resend to the original topic 3. Track real success/failure counts per message -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
