Copilot commented on code in PR #424:
URL:
https://github.com/apache/rocketmq-dashboard/pull/424#discussion_r3543451812
##########
src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImpl.java:
##########
@@ -196,12 +200,63 @@ public int compare(MessageView o1, MessageView o2) {
@Override
public List<MessageTrack> messageTrackDetail(MessageExt msg) {
+ List<MessageTrack> 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.
+ // 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 container differs from the
broker),
+ // the address comparison silently fails and every message is
incorrectly
+ // reported as NOT_CONSUME_YET even though it has already been
consumed.
+ //
+ // The fallback below re-checks the consumer offset directly, matching
+ // only on topic + queueId + offset and skipping the fragile address
+ // comparison. See:
https://github.com/apache/rocketmq-dashboard/issues/380
+ if (messageTracks != null) {
+ for (MessageTrack track : messageTracks) {
+ if (track.getTrackType() == TrackType.NOT_CONSUME_YET) {
+ if (isConsumedByGroup(msg, track.getConsumerGroup())) {
+ track.setTrackType(TrackType.CONSUMED);
+ }
+ }
+ }
+ }
+
+ return messageTracks;
+ }
+
+ /**
+ * Independently verify whether a message has been consumed by the given
+ * consumer group. Unlike {@link
org.apache.rocketmq.tools.admin.DefaultMQAdminExt#consumed},
+ * this method does NOT compare broker addresses — it matches solely on
+ * topic, queueId and consumer offset, which avoids false negatives caused
+ * by hostname/IP mismatches between the broker registry and store host.
+ */
+ private boolean isConsumedByGroup(MessageExt msg, String consumerGroup) {
+ try {
+ ConsumeStats stats = mqAdminExt.examineConsumeStats(consumerGroup);
+ if (stats == null || stats.getOffsetTable() == null) {
+ return false;
+ }
+ for (Map.Entry<MessageQueue, OffsetWrapper> entry :
stats.getOffsetTable().entrySet()) {
+ MessageQueue mq = entry.getKey();
+ if (mq.getTopic().equals(msg.getTopic()) && mq.getQueueId() ==
msg.getQueueId()) {
+ if (entry.getValue().getConsumerOffset() >
msg.getQueueOffset()) {
+ return true;
+ }
+ }
Review Comment:
isConsumedByGroup() matches ConsumeStats entries only by topic + queueId. In
RocketMQ, queueId is scoped per broker, so topics with multiple brokers can
have multiple MessageQueue entries with the same queueId. This can incorrectly
upgrade NOT_CONSUME_YET → CONSUMED based on a different broker’s queue offsets.
Also, entry.getValue() can be null, which would NPE.
##########
src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImpl.java:
##########
@@ -196,12 +200,63 @@ public int compare(MessageView o1, MessageView o2) {
@Override
public List<MessageTrack> messageTrackDetail(MessageExt msg) {
+ List<MessageTrack> 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.
+ // 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 container differs from the
broker),
+ // the address comparison silently fails and every message is
incorrectly
+ // reported as NOT_CONSUME_YET even though it has already been
consumed.
+ //
+ // The fallback below re-checks the consumer offset directly, matching
+ // only on topic + queueId + offset and skipping the fragile address
+ // comparison. See:
https://github.com/apache/rocketmq-dashboard/issues/380
+ if (messageTracks != null) {
+ for (MessageTrack track : messageTracks) {
+ if (track.getTrackType() == TrackType.NOT_CONSUME_YET) {
+ if (isConsumedByGroup(msg, track.getConsumerGroup())) {
+ track.setTrackType(TrackType.CONSUMED);
+ }
+ }
+ }
+ }
Review Comment:
messageTrackDetail() will call examineConsumeStats once per NOT_CONSUME_YET
track (via isConsumedByGroup). For messages with many consumer groups this
becomes multiple network round-trips per request and can noticeably slow down
the track-detail page. Cache the consumed check per consumer group within this
method so each group is examined at most once.
--
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]
