diff --git a/src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImpl.java b/src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImpl.java index 69a2b2ca..84b5e49a 100644 --- a/src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImpl.java +++ b/src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImpl.java @@ -37,6 +37,7 @@ import org.apache.rocketmq.common.message.MessageClientIDSetter; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.common.utils.NetworkUtil; import org.apache.rocketmq.dashboard.config.RMQConfigure; import org.apache.rocketmq.dashboard.exception.ServiceException; import org.apache.rocketmq.dashboard.model.MessagePage; @@ -48,11 +49,16 @@ import org.apache.rocketmq.dashboard.service.MessageService; import org.apache.rocketmq.dashboard.support.AutoCloseConsumerWrapper; import org.apache.rocketmq.remoting.RPCHook; +import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats; +import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper; import org.apache.rocketmq.remoting.protocol.body.Connection; import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult; import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection; +import org.apache.rocketmq.remoting.protocol.route.BrokerData; +import org.apache.rocketmq.remoting.protocol.route.TopicRouteData; import org.apache.rocketmq.tools.admin.MQAdminExt; import org.apache.rocketmq.tools.admin.api.MessageTrack; +import org.apache.rocketmq.tools.admin.api.TrackType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -65,7 +71,9 @@ import java.util.Collections; import java.util.Comparator; import java.util.Date; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -196,12 +204,152 @@ public int compare(MessageView o1, MessageView o2) { @Override public List messageTrackDetail(MessageExt msg) { + List messageTracks; try { - return mqAdminExt.messageTrackDetail(msg); + messageTracks = mqAdminExt.messageTrackDetail(msg); } catch (Exception e) { logger.error("op=messageTrackDetailError", e); return Collections.emptyList(); } + + // Re-verify tracks marked as NOT_CONSUME_YET. + // + // Root cause: the underlying consumed() method in DefaultMQAdminExtImpl + // requires an exact broker-address match between msg.getStoreHost() and + // the broker's registered address. When the broker registers with a + // hostname (or when DNS resolution inside the dashboard JVM differs from + // the broker), the address comparison silently fails and every message is + // incorrectly reported as NOT_CONSUME_YET even though it has been consumed. + // + // The fallback below re-checks the consumer offset directly. To avoid + // cross-broker false positives in multi-broker clusters (where broker-a + // and broker-b both have queueId=0 for the same topic with independent + // offsets), we first resolve msg.getStoreHost() to a brokerName via the + // topic route data, then only compare offsets for that brokerName. + // + // Conservative policy: if the brokerName cannot be determined (route info + // unavailable or no address matches), the track remains NOT_CONSUME_YET — + // we prefer preserving the original false negative over introducing a new + // false positive. + // + // See: https://github.com/apache/rocketmq-dashboard/issues/380 + if (messageTracks != null && !messageTracks.isEmpty()) { + // Cache ConsumeStats per consumer group within this request to avoid + // duplicate examineConsumeStats RPCs when multiple tracks share a group. + Map statsCache = new HashMap<>(); + for (MessageTrack track : messageTracks) { + if (track.getTrackType() == TrackType.NOT_CONSUME_YET) { + String group = track.getConsumerGroup(); + ConsumeStats stats = statsCache.computeIfAbsent(group, g -> { + try { + return mqAdminExt.examineConsumeStats(g); + } catch (Exception e) { + logger.warn("op=examineConsumeStatsError, group={}", g, e); + return null; + } + }); + if (stats != null && isConsumedByGroup(msg, stats)) { + track.setTrackType(TrackType.CONSUMED); + } + } + } + } + + return messageTracks; + } + + /** + * Independently verify whether a message has been consumed by the given + * consumer group, using pre-fetched {@link ConsumeStats}. + * + *

Unlike {@link org.apache.rocketmq.tools.admin.DefaultMQAdminExt#consumed}, + * this method does NOT rely on an exact broker-address string comparison. + * Instead it: + *

    + *
  1. Resolves the message's store host to a brokerName via topic route + * data, performing a loose IP-level match against all registered + * broker addresses.
    2. + *
  2. Only compares consumer offsets for {@link MessageQueue}s whose + * brokerName, topic, and queueId all match, preventing cross-broker + * false positives in multi-broker deployments.
    3. + *
+ * + * @param msg the message to verify + * @param stats pre-fetched ConsumeStats for the consumer group + * @return {@code true} only if the consumer offset for the matching broker + * has advanced past the message's queue offset; {@code false} if no + * matching broker is found or the offset has not been reached + */ + private boolean isConsumedByGroup(MessageExt msg, ConsumeStats stats) { + if (stats == null || stats.getOffsetTable() == null) { + return false; + } + + String targetBrokerName = resolveBrokerName(msg); + if (targetBrokerName == null) { + // Cannot determine which broker holds this message. + // Conservative: do not upgrade, leave NOT_CONSUME_YET. + return false; + } + + for (Map.Entry entry : stats.getOffsetTable().entrySet()) { + MessageQueue mq = entry.getKey(); + if (mq.getBrokerName().equals(targetBrokerName) + && mq.getTopic().equals(msg.getTopic()) + && mq.getQueueId() == msg.getQueueId()) { + if (entry.getValue().getConsumerOffset() > msg.getQueueOffset()) { + return true; + } + } + } + return false; + } + + /** + * Resolve the brokerName that owns the given message by matching + * {@code msg.getStoreHost()} against the topic's route data. + * + *

The match is performed at the IP level: both the store host and the + * registered broker addresses are normalised to their IP components before + * comparison, so a broker that registers with a hostname whose DNS resolves + * to the same IP as the store host will still match. + * + * @return the matching brokerName, or {@code null} if no match is found + */ + private String resolveBrokerName(MessageExt msg) { + try { + TopicRouteData routeData = mqAdminExt.examineTopicRouteInfo(msg.getTopic()); + if (routeData == null || routeData.getBrokerDatas() == null) { + return null; + } + + List brokerDatas = routeData.getBrokerDatas(); + + // Short-circuit: single-broker deployment — no address matching needed. + // This covers the most common deployment (dev / test / small-scale prod) + // and is immune to DNS/hostname/NAT mismatches that would otherwise + // cause the IP-level match below to fail silently. + if (brokerDatas.size() == 1) { + return brokerDatas.get(0).getBrokerName(); + } + + String storeHostIp = NetworkUtil.socketAddress2String(msg.getStoreHost()).split(":")[0]; + + for (BrokerData brokerData : brokerDatas) { + if (brokerData.getBrokerAddrs() == null) { + continue; + } + for (String brokerAddr : brokerData.getBrokerAddrs().values()) { + String brokerIp = NetworkUtil.convert2IpString(brokerAddr).split(":")[0]; + if (storeHostIp.equals(brokerIp)) { + return brokerData.getBrokerName(); + } + } + } + } catch (Exception e) { + logger.warn("op=resolveBrokerNameError, topic={}", msg.getTopic(), e); + } + return null; } diff --git a/src/test/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImplTest.java b/src/test/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImplTest.java index e3280538..fc5e61fb 100644 --- a/src/test/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImplTest.java +++ b/src/test/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImplTest.java @@ -34,11 +34,16 @@ import org.apache.rocketmq.dashboard.model.QueueOffsetInfo; import org.apache.rocketmq.dashboard.support.AutoCloseConsumerWrapper; import org.apache.rocketmq.remoting.RPCHook; +import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats; +import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper; import org.apache.rocketmq.remoting.protocol.body.Connection; import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult; import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection; +import org.apache.rocketmq.remoting.protocol.route.BrokerData; +import org.apache.rocketmq.remoting.protocol.route.TopicRouteData; import org.apache.rocketmq.tools.admin.MQAdminExt; import org.apache.rocketmq.tools.admin.api.MessageTrack; +import org.apache.rocketmq.tools.admin.api.TrackType; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -48,11 +53,15 @@ import org.mockito.junit.MockitoJUnitRunner; import java.lang.reflect.Method; +import java.net.InetAddress; +import java.net.InetSocketAddress; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; import static org.junit.Assert.assertEquals; @@ -441,10 +450,389 @@ private MessageExt createMessageExt(String msgId, String topic, String body, lon msg.setTopic(topic); msg.setBody(body.getBytes()); msg.setStoreTimestamp(storeTimestamp); + msg.setBornHost(new InetSocketAddress("127.0.0.1", 0)); + msg.setStoreHost(new InetSocketAddress("192.168.1.100", 10911)); return msg; } + /** + * Create a TopicRouteData with a single broker whose address matches + * the default storeHost (192.168.1.100:10911) used by createMessageExt. + */ + private TopicRouteData createSingleBrokerRouteData(String brokerName) { + TopicRouteData routeData = new TopicRouteData(); + HashMap brokerAddrs = new HashMap<>(); + brokerAddrs.put(0L, "192.168.1.100:10911"); + List brokerDatas = Collections.singletonList( + new BrokerData("test-cluster", brokerName, brokerAddrs)); + routeData.setBrokerDatas(brokerDatas); + return routeData; + } + + /** + * Create a TopicRouteData with two brokers, only one of which matches + * the default storeHost IP. Used to verify cross-broker isolation. + */ + private TopicRouteData createMultiBrokerRouteData() { + TopicRouteData routeData = new TopicRouteData(); + List brokerDatas = new ArrayList<>(); + + HashMap addrsA = new HashMap<>(); + addrsA.put(0L, "192.168.1.100:10911"); + brokerDatas.add(new BrokerData("test-cluster", "broker-a", addrsA)); + + HashMap addrsB = new HashMap<>(); + addrsB.put(0L, "192.168.2.200:10911"); + brokerDatas.add(new BrokerData("test-cluster", "broker-b", addrsB)); + + routeData.setBrokerDatas(brokerDatas); + return routeData; + } + private PullResult createPullResult(PullStatus status, List msgFoundList, long nextBeginOffset, long minOffset) { return new PullResult(status, nextBeginOffset, minOffset, minOffset + msgFoundList.size(), msgFoundList); } + + // ==================== Tests for issue #380: NOT_CONSUME_YET false positive ==================== + + @Test + public void testMessageTrackDetail_NotConsumeYetCorrectedToConsumed() throws Exception { + // Simulate the scenario from issue #380: + // DefaultMQAdminExtImpl.consumed() returns false due to broker address mismatch, + // resulting in NOT_CONSUME_YET, but the message was actually consumed. + MessageExt msg = createMessageExt(MSG_ID, TOPIC, "body", System.currentTimeMillis()); + msg.setQueueId(2); + msg.setQueueOffset(100); + + // messageTrackDetail from MQAdminExt returns NOT_CONSUME_YET + MessageTrack track = new MessageTrack(); + track.setConsumerGroup(CONSUMER_GROUP); + track.setTrackType(TrackType.NOT_CONSUME_YET); + + when(mqAdminExt.messageTrackDetail(any(MessageExt.class))) + .thenReturn(Collections.singletonList(track)); + + // Topic route data maps the message's store host IP to broker-a + when(mqAdminExt.examineTopicRouteInfo(TOPIC)) + .thenReturn(createSingleBrokerRouteData("broker-a")); + + // ConsumeStats shows consumerOffset (200) > msg.queueOffset (100) → consumed + ConsumeStats stats = new ConsumeStats(); + MessageQueue mq = new MessageQueue(TOPIC, "broker-a", 2); + OffsetWrapper offsetWrapper = new OffsetWrapper(); + offsetWrapper.setConsumerOffset(200); + offsetWrapper.setBrokerOffset(250); + Map offsetTable = new HashMap<>(); + offsetTable.put(mq, offsetWrapper); + stats.setOffsetTable(offsetTable); + + when(mqAdminExt.examineConsumeStats(CONSUMER_GROUP)).thenReturn(stats); + + // Execute + List result = messageService.messageTrackDetail(msg); + + // Verify NOT_CONSUME_YET was corrected to CONSUMED + assertNotNull(result); + assertEquals(1, result.size()); + assertEquals(TrackType.CONSUMED, result.get(0).getTrackType()); + } + + @Test + public void testMessageTrackDetail_NotConsumeYetRemainsWhenNotConsumed() throws Exception { + // When the consumer offset is actually behind the message offset, + // NOT_CONSUME_YET should remain unchanged. + MessageExt msg = createMessageExt(MSG_ID, TOPIC, "body", System.currentTimeMillis()); + msg.setQueueId(0); + msg.setQueueOffset(500); + + MessageTrack track = new MessageTrack(); + track.setConsumerGroup(CONSUMER_GROUP); + track.setTrackType(TrackType.NOT_CONSUME_YET); + + when(mqAdminExt.messageTrackDetail(any(MessageExt.class))) + .thenReturn(Collections.singletonList(track)); + + when(mqAdminExt.examineTopicRouteInfo(TOPIC)) + .thenReturn(createSingleBrokerRouteData("broker-a")); + + // consumerOffset (100) < msg.queueOffset (500) → genuinely not consumed yet + ConsumeStats stats = new ConsumeStats(); + MessageQueue mq = new MessageQueue(TOPIC, "broker-a", 0); + OffsetWrapper offsetWrapper = new OffsetWrapper(); + offsetWrapper.setConsumerOffset(100); + offsetWrapper.setBrokerOffset(600); + Map offsetTable = new HashMap<>(); + offsetTable.put(mq, offsetWrapper); + stats.setOffsetTable(offsetTable); + + when(mqAdminExt.examineConsumeStats(CONSUMER_GROUP)).thenReturn(stats); + + // Execute + List result = messageService.messageTrackDetail(msg); + + // Verify NOT_CONSUME_YET remains + assertNotNull(result); + assertEquals(1, result.size()); + assertEquals(TrackType.NOT_CONSUME_YET, result.get(0).getTrackType()); + } + + @Test + public void testMessageTrackDetail_CrossBrokerNoFalsePositive() throws Exception { + // Multi-broker scenario: message is stored on broker-a queue-0, but + // broker-b also has queue-0 for the same topic with a higher consumer + // offset. The fix must NOT match across brokers. + MessageExt msg = createMessageExt(MSG_ID, TOPIC, "body", System.currentTimeMillis()); + msg.setQueueId(0); + msg.setQueueOffset(1000); + + MessageTrack track = new MessageTrack(); + track.setConsumerGroup(CONSUMER_GROUP); + track.setTrackType(TrackType.NOT_CONSUME_YET); + + when(mqAdminExt.messageTrackDetail(any(MessageExt.class))) + .thenReturn(Collections.singletonList(track)); + + // Route data: broker-a at 192.168.1.100 (matches storeHost), + // broker-b at 192.168.2.200 (different IP) + when(mqAdminExt.examineTopicRouteInfo(TOPIC)) + .thenReturn(createMultiBrokerRouteData()); + + // ConsumeStats has entries for both brokers' queue-0. + // broker-a (the real owner): consumerOffset 500 < msg offset 1000 → not consumed + // broker-b (wrong broker): consumerOffset 5000 > msg offset 1000 → would be false positive + ConsumeStats stats = new ConsumeStats(); + Map offsetTable = new HashMap<>(); + + MessageQueue mqA = new MessageQueue(TOPIC, "broker-a", 0); + OffsetWrapper owA = new OffsetWrapper(); + owA.setConsumerOffset(500); + owA.setBrokerOffset(1000); + offsetTable.put(mqA, owA); + + MessageQueue mqB = new MessageQueue(TOPIC, "broker-b", 0); + OffsetWrapper owB = new OffsetWrapper(); + owB.setConsumerOffset(5000); + owB.setBrokerOffset(5000); + offsetTable.put(mqB, owB); + + stats.setOffsetTable(offsetTable); + when(mqAdminExt.examineConsumeStats(CONSUMER_GROUP)).thenReturn(stats); + + // Execute + List result = messageService.messageTrackDetail(msg); + + // Verify: must remain NOT_CONSUME_YET — broker-b's high offset must NOT leak + assertNotNull(result); + assertEquals(1, result.size()); + assertEquals(TrackType.NOT_CONSUME_YET, result.get(0).getTrackType()); + } + + @Test + public void testMessageTrackDetail_BrokerNameUnresolvableStaysNotConsumeYet() throws Exception { + // Multi-broker route where NO broker's address matches msg.getStoreHost(). + // The short-circuit does not apply (>1 broker), IP matching fails on all + // brokers, so resolveBrokerName returns null → conservative NOT_CONSUME_YET. + MessageExt msg = createMessageExt(MSG_ID, TOPIC, "body", System.currentTimeMillis()); + msg.setQueueId(0); + msg.setQueueOffset(100); + + MessageTrack track = new MessageTrack(); + track.setConsumerGroup(CONSUMER_GROUP); + track.setTrackType(TrackType.NOT_CONSUME_YET); + + when(mqAdminExt.messageTrackDetail(any(MessageExt.class))) + .thenReturn(Collections.singletonList(track)); + + // Two brokers, neither matching the storeHost IP (192.168.1.100) + TopicRouteData routeData = new TopicRouteData(); + List brokerDatas = new ArrayList<>(); + HashMap addrsX = new HashMap<>(); + addrsX.put(0L, "10.0.0.1:10911"); + brokerDatas.add(new BrokerData("test-cluster", "broker-x", addrsX)); + HashMap addrsY = new HashMap<>(); + addrsY.put(0L, "10.0.0.2:10911"); + brokerDatas.add(new BrokerData("test-cluster", "broker-y", addrsY)); + routeData.setBrokerDatas(brokerDatas); + when(mqAdminExt.examineTopicRouteInfo(TOPIC)).thenReturn(routeData); + + // ConsumeStats has matching topic+queueId entries for both brokers, + // but since resolveBrokerName can't determine which broker holds the + // message (no IP match), neither entry should be used. + ConsumeStats stats = new ConsumeStats(); + MessageQueue mqX = new MessageQueue(TOPIC, "broker-x", 0); + MessageQueue mqY = new MessageQueue(TOPIC, "broker-y", 0); + OffsetWrapper owX = new OffsetWrapper(); + owX.setConsumerOffset(9999); + OffsetWrapper owY = new OffsetWrapper(); + owY.setConsumerOffset(8888); + Map offsetTable = new HashMap<>(); + offsetTable.put(mqX, owX); + offsetTable.put(mqY, owY); + stats.setOffsetTable(offsetTable); + when(mqAdminExt.examineConsumeStats(CONSUMER_GROUP)).thenReturn(stats); + + // Execute + List result = messageService.messageTrackDetail(msg); + + // Verify: NOT_CONSUME_YET preserved — cannot determine broker, so conservative + assertNotNull(result); + assertEquals(1, result.size()); + assertEquals(TrackType.NOT_CONSUME_YET, result.get(0).getTrackType()); + } + + @Test + public void testMessageTrackDetail_SingleBrokerShortCircuitWithMismatchedIp() throws Exception { + // Single-broker deployment where the broker registers with a hostname/IP + // that does NOT match msg.getStoreHost(). The single-broker short-circuit + // should still resolve the brokerName, allowing the fallback to work — + // this is the exact scenario reported in issue #380. + MessageExt msg = createMessageExt(MSG_ID, TOPIC, "body", System.currentTimeMillis()); + msg.setQueueId(2); + msg.setQueueOffset(100); + + MessageTrack track = new MessageTrack(); + track.setConsumerGroup(CONSUMER_GROUP); + track.setTrackType(TrackType.NOT_CONSUME_YET); + + when(mqAdminExt.messageTrackDetail(any(MessageExt.class))) + .thenReturn(Collections.singletonList(track)); + + // Single broker with a non-matching address (DNS/hostname mismatch scenario) + TopicRouteData routeData = new TopicRouteData(); + HashMap addrs = new HashMap<>(); + addrs.put(0L, "broker-a-hostname:10911"); // does NOT match 192.168.1.100 + routeData.setBrokerDatas(Collections.singletonList( + new BrokerData("test-cluster", "broker-a", addrs))); + when(mqAdminExt.examineTopicRouteInfo(TOPIC)).thenReturn(routeData); + + // Consumer offset has advanced past the message + ConsumeStats stats = new ConsumeStats(); + MessageQueue mq = new MessageQueue(TOPIC, "broker-a", 2); + OffsetWrapper ow = new OffsetWrapper(); + ow.setConsumerOffset(200); + Map offsetTable = new HashMap<>(); + offsetTable.put(mq, ow); + stats.setOffsetTable(offsetTable); + when(mqAdminExt.examineConsumeStats(CONSUMER_GROUP)).thenReturn(stats); + + // Execute + List result = messageService.messageTrackDetail(msg); + + // Verify: corrected to CONSUMED despite IP mismatch, because single-broker + // short-circuit resolved the brokerName without address matching. + assertNotNull(result); + assertEquals(1, result.size()); + assertEquals(TrackType.CONSUMED, result.get(0).getTrackType()); + } + + @Test + public void testMessageTrackDetail_ConsumedTrackUnchanged() throws Exception { + // Tracks that are already CONSUMED should not be re-verified + MessageExt msg = createMessageExt(MSG_ID, TOPIC, "body", System.currentTimeMillis()); + + MessageTrack track = new MessageTrack(); + track.setConsumerGroup(CONSUMER_GROUP); + track.setTrackType(TrackType.CONSUMED); + + when(mqAdminExt.messageTrackDetail(any(MessageExt.class))) + .thenReturn(Collections.singletonList(track)); + + // Execute + List result = messageService.messageTrackDetail(msg); + + // Verify CONSUMED is unchanged, and examineConsumeStats was never called + assertNotNull(result); + assertEquals(1, result.size()); + assertEquals(TrackType.CONSUMED, result.get(0).getTrackType()); + verify(mqAdminExt, never()).examineConsumeStats(anyString()); + } + + @Test + public void testMessageTrackDetail_NotOnlineTrackUnchanged() throws Exception { + // NOT_ONLINE tracks should not be re-verified + MessageExt msg = createMessageExt(MSG_ID, TOPIC, "body", System.currentTimeMillis()); + + MessageTrack track = new MessageTrack(); + track.setConsumerGroup(CONSUMER_GROUP); + track.setTrackType(TrackType.NOT_ONLINE); + + when(mqAdminExt.messageTrackDetail(any(MessageExt.class))) + .thenReturn(Collections.singletonList(track)); + + // Execute + List result = messageService.messageTrackDetail(msg); + + assertNotNull(result); + assertEquals(TrackType.NOT_ONLINE, result.get(0).getTrackType()); + verify(mqAdminExt, never()).examineConsumeStats(anyString()); + } + + @Test + public void testMessageTrackDetail_ExamineConsumeStatsThrowsException() throws Exception { + // When examineConsumeStats throws, the original NOT_CONSUME_YET should be preserved + MessageExt msg = createMessageExt(MSG_ID, TOPIC, "body", System.currentTimeMillis()); + msg.setQueueId(0); + msg.setQueueOffset(100); + + MessageTrack track = new MessageTrack(); + track.setConsumerGroup(CONSUMER_GROUP); + track.setTrackType(TrackType.NOT_CONSUME_YET); + + when(mqAdminExt.messageTrackDetail(any(MessageExt.class))) + .thenReturn(Collections.singletonList(track)); + when(mqAdminExt.examineTopicRouteInfo(TOPIC)) + .thenReturn(createSingleBrokerRouteData("broker-a")); + when(mqAdminExt.examineConsumeStats(CONSUMER_GROUP)) + .thenThrow(new RuntimeException("Connection refused")); + + // Execute — should not throw + List result = messageService.messageTrackDetail(msg); + + assertNotNull(result); + assertEquals(1, result.size()); + assertEquals(TrackType.NOT_CONSUME_YET, result.get(0).getTrackType()); + } + + @Test + public void testMessageTrackDetail_GroupLevelStatsCache() throws Exception { + // Verify that when multiple NOT_CONSUME_YET tracks share the same + // consumer group, examineConsumeStats is called only once (cached). + MessageExt msg = createMessageExt(MSG_ID, TOPIC, "body", System.currentTimeMillis()); + msg.setQueueId(0); + msg.setQueueOffset(100); + + MessageTrack track1 = new MessageTrack(); + track1.setConsumerGroup(CONSUMER_GROUP); + track1.setTrackType(TrackType.NOT_CONSUME_YET); + + MessageTrack track2 = new MessageTrack(); + track2.setConsumerGroup(CONSUMER_GROUP); // same group + track2.setTrackType(TrackType.NOT_CONSUME_YET); + + when(mqAdminExt.messageTrackDetail(any(MessageExt.class))) + .thenReturn(Arrays.asList(track1, track2)); + + when(mqAdminExt.examineTopicRouteInfo(TOPIC)) + .thenReturn(createSingleBrokerRouteData("broker-a")); + + ConsumeStats stats = new ConsumeStats(); + MessageQueue mq = new MessageQueue(TOPIC, "broker-a", 0); + OffsetWrapper ow = new OffsetWrapper(); + ow.setConsumerOffset(200); + Map offsetTable = new HashMap<>(); + offsetTable.put(mq, ow); + stats.setOffsetTable(offsetTable); + + when(mqAdminExt.examineConsumeStats(CONSUMER_GROUP)).thenReturn(stats); + + // Execute + List result = messageService.messageTrackDetail(msg); + + // Both tracks should be corrected, but examineConsumeStats called only once + assertNotNull(result); + assertEquals(2, result.size()); + assertEquals(TrackType.CONSUMED, result.get(0).getTrackType()); + assertEquals(TrackType.CONSUMED, result.get(1).getTrackType()); + verify(mqAdminExt, times(1)).examineConsumeStats(CONSUMER_GROUP); + } }