This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new e45708056 fix(group): preview offset resets for offline consumer
groups (#5291)
e45708056 is described below
commit e457080569850efb7d7de374bbaa30a24701536f
Author: Apulupie <[email protected]>
AuthorDate: Tue Oct 27 15:46:50 2026 +0800
fix(group): preview offset resets for offline consumer groups (#5291)
* fix(group): preview the offset reset from the topic route for stopped
groups
The preview called examineConsumeStats and, on CONSUMER_NOT_ONLINE,
returned an
empty preview with allowReset=false - so a stopped group could never be
previewed
even though POST /api/groups/reset-offset (which does not validate a
preview)
reset it regardless. In rocketmq-tools 5.5.0, resetOffsetNew calls
resetOffsetByTimestamp(force=true) and, on response code 206, falls back to
resetOffsetByTimestampOld(force=true), which reads the broker offset store
and
writes searchOffset() for each queue of the topic route. The preview now
walks
that same route and issues the same per-queue search, so an offline group
sees
the targets the reset will actually commit.
Broadcast groups (213) and groups that never consumed (17 on
%RETRY%<group>) are
still blocked up front and keep returning the empty, non-resettable preview.
* fix(group): keep the offline reset preview free of the -1 sentinel
Current offsets genuinely cannot be read for a group that is not connected,
so the
lag and delta fields carry the same unknown sentinels the online path
already
uses. They must not reach user-facing prose: the tool plan read "projected
lag
changes from -1 to 500", and the Chinese description column echoed an
English
sentence built around a delta. Both now render an explicit unavailable
state, and
the aggregate net-delta field renders '-' when any queue had no current
offset,
matching what the per-queue column already did.
Disclosure - this unblocks two gates derived from the preview.
rmq.group.reset_offset
is riskLevel L2 with permission group:write and requires a confirm_token
that only a
successful preview() can mint, and the UI's confirm-reset button keys off
the same
allowReset, so an offline group could not confirm a reset through either
path before.
No new capability is granted: the REST endpoint never validated a preview,
and the
server-side execution path has used resetOffsetNew since #3310. The gates
themselves
are unchanged - !complete || !allowReset still throws 409, and any queue
that fails
to preview still marks the whole preview incomplete. The plan gains a
warning that
current offsets are unavailable so deltas cannot be computed.
Known residual case, not introduced here: a broadcast group that is also
offline
raises 206 before the broadcast check can run, so it reaches the offline
preview and
is let through. resetOffsetNew then writes broker-side offsets that
broadcast
consumers never read - identical to `mqadmin resetOffsetByTime` on the same
input,
and already reachable through the REST endpoint. The subscription mode
lives on the
client while a group is offline, so the server cannot rule this out; the
warning
hands the uncertainty to whoever confirms the reset.
---
.../handler/group/GroupResetOffsetToolHandler.java | 19 +-
.../provider/apache/RocketMQAdminClientImpl.java | 195 ++++++++++++++++++++-
.../group/GroupResetOffsetToolHandlerTest.java | 62 +++++++
.../apache/RocketMQAdminClientImplTest.java | 111 ++++++++++++
web/src/i18n/translations.ts | 9 +
.../pages/instance/__tests__/ConsumerPage.test.tsx | 84 +++++++++
web/src/pages/instance/consumer.tsx | 43 +++--
7 files changed, 505 insertions(+), 18 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupResetOffsetToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupResetOffsetToolHandler.java
index 6b2c5f0bf..78fa26f03 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupResetOffsetToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupResetOffsetToolHandler.java
@@ -66,8 +66,8 @@ public class GroupResetOffsetToolHandler extends
MutationToolHandler<GroupResetO
.before(OffsetState.current(input, preview))
.after(OffsetState.proposed(input, preview, timestamp))
.impact("Moves offsets for " + preview.getQueueCount()
- + " queues; projected lag changes from " +
preview.getCurrentTotalLag()
- + " to " + preview.getProjectedTotalLag() + ".")
+ + " queues; projected lag changes from " +
lagPhrase(preview.getCurrentTotalLag())
+ + " to " + lagPhrase(preview.getProjectedTotalLag()) +
".")
.warnings(warnings)
.build();
}
@@ -108,6 +108,12 @@ public class GroupResetOffsetToolHandler extends
MutationToolHandler<GroupResetO
}
}
+ /**
+ * One queue's offsets. {@code offset} and {@code lag} carry the {@code
-1} sentinel the preview
+ * reports for a value the broker could not determine — an offline group
has no current offset,
+ * so its "before" side is {@code -1} on both fields. Structured output
keeps the sentinel the
+ * way {@code GroupListItem} documents its own, and {@link #lagPhrase}
keeps it out of prose.
+ */
private record QueueState(String topic, String broker, int queueId, long
offset, long lag) {
static QueueState current(ResetConsumerOffsetQueuePreviewVO queue) {
@@ -120,4 +126,13 @@ public class GroupResetOffsetToolHandler extends
MutationToolHandler<GroupResetO
queue.getTargetOffset(), queue.getProjectedLag());
}
}
+
+ /**
+ * Renders a lag for the plan prose. The {@code -1} "cannot be determined"
sentinel must never
+ * be interpolated as a number, or the plan reads "projected lag changes
from -1 to 500";
+ * {@code unavailable} is the wording {@code GroupDetailToolHandler}
already uses for it.
+ */
+ private static String lagPhrase(long lag) {
+ return lag < 0 ? "unavailable" : Long.toString(lag);
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
index e7fb3378b..ca6016f58 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
@@ -26,6 +26,7 @@ import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicAttributes;
+import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
@@ -34,6 +35,8 @@ import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.QueueData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.MqClientPool;
@@ -91,6 +94,12 @@ public class RocketMQAdminClientImpl implements AdminClient {
private static final int MAX_MESSAGE_SIZE = 4 * 1024 * 1024; // 4 MB
default broker limit
private static final long RESET_OFFSET_PREVIEW_TIMEOUT_MILLIS = 3_000L;
+ /**
+ * Warning every offset-reset preview path carries when the broker
reported no consume offset:
+ * the group is not connected. The paths differ in what they can still
compute from it.
+ */
+ private static final String OFFLINE_GROUP_PREVIEW_WARNING =
+ "Consumer group is not online and no consume offset data is
available";
private static final String RISK_INFO = "INFO";
private static final String RISK_WARNING = "WARNING";
private static final String RISK_ERROR = "ERROR";
@@ -1070,14 +1079,173 @@ public class RocketMQAdminClientImpl implements
AdminClient {
.queues(queues)
.build();
} catch (Exception e) {
- if (isConsumerNotOnline(e)) {
+ if (isBroadcastOrNeverConsumedGroup(e)) {
+ // Broadcast groups keep offsets per consumer and a group with
no %RETRY% route
+ // has never consumed: there is nothing meaningful to preview,
so keep blocking.
return emptyResetOffsetPreview(instanceId, name, timestamp,
topic,
- "Consumer group is not online and no consume offset
data is available");
+ OFFLINE_GROUP_PREVIEW_WARNING);
+ }
+ if (isConsumerNotOnline(e)) {
+ // CONSUMER_NOT_ONLINE only means the broker-side offset table
came back empty for
+ // a group that is not connected; resetOffsetNew still applies
to it, so preview
+ // from the topic route instead of permanently blocking the
reset.
+ return offlineResetOffsetPreview(instanceId, admin, name,
timestamp, topic);
}
throw new BusinessException(500, "Failed to preview reset offset:
" + e.getMessage());
}
}
+ /**
+ * Grades the two "no data at all" exceptions {@code examineConsumeStats}
raises as still
+ * blocking, so they do not reach {@link #offlineResetOffsetPreview}.
+ *
+ * <p>{@code BROADCAST_CONSUMPTION} marks a broadcast group: its offsets
live in each client,
+ * so a broker-side write would not reset anything the group reads. {@code
TOPIC_NOT_EXIST} on
+ * {@code %RETRY%<group>} means rocketmq-tools never resolved the group,
i.e. it has never
+ * consumed, and creating broker-side offsets for it is not a preview.
+ */
+ private static boolean isBroadcastOrNeverConsumedGroup(Exception
exception) {
+ if (MqResponseCodes.hasResponseCode(exception,
ResponseCode.BROADCAST_CONSUMPTION)) {
+ return true;
+ }
+ String message = exception.getMessage();
+ return MqResponseCodes.hasResponseCode(exception,
ResponseCode.TOPIC_NOT_EXIST)
+ && message != null && message.contains("%RETRY%");
+ }
+
+ /**
+ * Preview for a group that is not connected, mirroring what {@code
resetOffsetNew} will do to
+ * it. {@code MQAdminExt.resetOffsetNew(group, topic, timestamp)}
(rocketmq-tools 5.5.0) calls
+ * {@code resetOffsetByTimestamp} with {@code force = true} and, on
+ * {@code MQClientException(CONSUMER_NOT_ONLINE)}, falls back to
+ * {@code resetOffsetByTimestampOld(group, topic, timestamp, true)}. That
fallback reads the
+ * broker offset store rather than the live consumers and, when it finds
no consume stats for
+ * the topic, enumerates {@code queueData.getReadQueueNums()} queues of
the topic route and
+ * writes {@code searchOffset(brokerAddr, queue, timestamp)} to each one
unconditionally
+ * ({@code force}). This preview therefore walks the same route and the
same per-queue search,
+ * so an offline group sees the targets the reset will actually commit.
+ *
+ * <p>Current consumer offsets are genuinely unavailable offline, so the
lag and delta fields
+ * carry the {@link ConsumerLagResolver#UNKNOWN} / neutral sentinels the
online path already
+ * uses and the preview adds an explicit offline warning; presentation
layers render those
+ * sentinels as "unavailable" instead of as numbers. Falls back to the
empty preview when the
+ * topic route is unavailable too.
+ */
+ private ResetConsumerOffsetPreviewVO offlineResetOffsetPreview(String
instanceId, MQAdminExt admin,
+ String
name, long timestamp,
+ String
topic) {
+ List<ResetConsumerOffsetQueuePreviewVO> queues = new ArrayList<>();
+ try {
+ TopicRouteData route = admin.examineTopicRouteInfo(topic);
+ if (route != null && route.getQueueDatas() != null) {
+ for (QueueData queueData : route.getQueueDatas()) {
+ // resetOffsetByTimestampOld enumerates the read queues of
every routed broker;
+ // a queue the route does not expose for reading is never
consumed from, so the
+ // reset cannot move it and the preview must not claim a
target for it.
+ if (queueData == null ||
!PermName.isReadable(queueData.getPerm())) {
+ continue;
+ }
+ for (int queueId = 0; queueId <
queueData.getReadQueueNums(); queueId++) {
+ queues.add(offlineResetOffsetQueue(admin,
+ new MessageQueue(topic,
queueData.getBrokerName(), queueId), timestamp));
+ }
+ }
+ }
+ } catch (Exception routeFailure) {
+ return emptyResetOffsetPreview(instanceId, name, timestamp, topic,
+ OFFLINE_GROUP_PREVIEW_WARNING
+ + "; the topic route is also unavailable: " +
routeFailure.getMessage());
+ }
+ if (queues.isEmpty()) {
+ return emptyResetOffsetPreview(instanceId, name, timestamp, topic,
+ OFFLINE_GROUP_PREVIEW_WARNING + "; no readable queues
found for topic " + topic);
+ }
+ queues.sort(Comparator
+ .comparing((ResetConsumerOffsetQueuePreviewVO queue) ->
+ queue.getBroker() == null ? "" :
queue.getBroker(),
+ String.CASE_INSENSITIVE_ORDER)
+
.thenComparingInt(ResetConsumerOffsetQueuePreviewVO::getQueueId));
+ long currentTotalLag = ConsumerLagResolver.UNKNOWN;
+ long projectedTotalLag = aggregateResetPreviewLag(queues, true);
+ // Delta is unavailable offline (no current offset), so no queue is
classified as a rewind
+ // or a fast-forward; the warnings only carry the unavailable-offset
and reset-extent notes.
+ List<String> warnings = new
ArrayList<>(buildResetOffsetPreviewWarnings(queues, 0, 0));
+ warnings.add(OFFLINE_GROUP_PREVIEW_WARNING
+ + "; current offsets are unavailable, so offset deltas cannot
be computed");
+ boolean complete = queues.stream().noneMatch(queue ->
RISK_ERROR.equals(queue.getRiskLevel()));
+ return ResetConsumerOffsetPreviewVO.builder()
+ .instanceId(instanceId)
+ .groupName(name)
+ .topic(topic)
+ .timestamp(timestamp)
+ .complete(complete)
+ .allowReset(complete)
+ .queueCount(queues.size())
+ .warningCount(warnings.size())
+ .rewindQueueCount(0)
+ .fastForwardQueueCount(0)
+ .currentTotalLag(currentTotalLag)
+ .projectedTotalLag(projectedTotalLag)
+ .totalOffsetDelta(0L)
+ .warnings(warnings)
+ .queues(queues)
+ .build();
+ }
+
+ /**
+ * Computes the target {@code resetOffsetNew} commits for one routed queue
of an offline group:
+ * {@code searchOffset(brokerAddr, queue, timestamp)}, clamped to the
retained range the same way
+ * {@link #previewResetOffsetQueue} clamps it. The broker maximum is the
broker offset (nothing
+ * is being consumed), so the projected lag is real; the current offset
and lag stay
+ * {@link ConsumerLagResolver#UNKNOWN} because no consumer reported them.
+ */
+ private ResetConsumerOffsetQueuePreviewVO
offlineResetOffsetQueue(MQAdminExt admin, MessageQueue queue,
+ long
timestamp) {
+ try {
+ long minOffset = admin.minOffset(queue);
+ long maxOffset = admin.maxOffset(queue);
+ String brokerAddr = resolveBrokerAddress(admin,
queue.getBrokerName());
+ long targetOffset = admin.searchOffset(brokerAddr,
queue.getTopic(), queue.getQueueId(),
+ timestamp, RESET_OFFSET_PREVIEW_TIMEOUT_MILLIS);
+ targetOffset = clampOffset(targetOffset, minOffset, maxOffset);
+ return ResetConsumerOffsetQueuePreviewVO.builder()
+ .topic(queue.getTopic())
+ .broker(queue.getBrokerName())
+ .queueId(queue.getQueueId())
+ .minOffset(minOffset)
+ .maxOffset(maxOffset)
+ .brokerOffset(maxOffset)
+ .consumerOffset(ConsumerLagResolver.UNKNOWN)
+ .targetOffset(targetOffset)
+ .currentLag(ConsumerLagResolver.UNKNOWN)
+ .projectedLag(resolveLag(maxOffset, targetOffset))
+ .offsetDelta(0L)
+ .riskLevel(resetOffsetRiskLevel(0, targetOffset,
minOffset, maxOffset))
+ // Never splice resetOffsetPreviewMessage in here: its
first clause states an
+ // offset delta, and the delta is unavailable offline,
which read as
+ // "moves the offset to Offset unchanged".
+ .message("Current offset unavailable (group offline);
target offset " + targetOffset
+ + resetOffsetExtentNote(targetOffset, minOffset,
maxOffset))
+ .build();
+ } catch (Exception e) {
+ return ResetConsumerOffsetQueuePreviewVO.builder()
+ .topic(queue.getTopic())
+ .broker(queue.getBrokerName())
+ .queueId(queue.getQueueId())
+ .minOffset(-1L)
+ .maxOffset(-1L)
+ .brokerOffset(-1L)
+ .consumerOffset(ConsumerLagResolver.UNKNOWN)
+ .targetOffset(-1L)
+ .currentLag(ConsumerLagResolver.UNKNOWN)
+ .projectedLag(ConsumerLagResolver.UNKNOWN)
+ .offsetDelta(0L)
+ .riskLevel(RISK_ERROR)
+ .message("Failed to preview queue offset: " +
e.getMessage())
+ .build();
+ }
+ }
+
private ResetConsumerOffsetQueuePreviewVO
previewResetOffsetQueue(MQAdminExt admin, MessageQueue queue,
OffsetWrapper wrapper, long timestamp) {
long brokerOffset = wrapper == null ? 0L : wrapper.getBrokerOffset();
@@ -1232,13 +1400,30 @@ public class RocketMQAdminClientImpl implements
AdminClient {
} else {
messages.add("Offset unchanged");
}
+ messages.addAll(resetOffsetExtentNotes(targetOffset, minOffset,
maxOffset));
+ return String.join("; ", messages);
+ }
+
+ /**
+ * Notes stating that the target sits on an end of the retained range.
Shared by the online
+ * message, whose first clause is the offset delta, and the offline
message, which has no delta
+ * to state and therefore must not borrow one.
+ */
+ private List<String> resetOffsetExtentNotes(long targetOffset, long
minOffset, long maxOffset) {
+ List<String> notes = new ArrayList<>();
if (targetOffset == minOffset) {
- messages.add("target is the minimum retained offset");
+ notes.add("target is the minimum retained offset");
}
if (targetOffset == maxOffset) {
- messages.add("target is the latest offset");
+ notes.add("target is the latest offset");
}
- return String.join("; ", messages);
+ return notes;
+ }
+
+ /** The extent notes as a message suffix; empty when the target is inside
the retained range. */
+ private String resetOffsetExtentNote(long targetOffset, long minOffset,
long maxOffset) {
+ List<String> notes = resetOffsetExtentNotes(targetOffset, minOffset,
maxOffset);
+ return notes.isEmpty() ? "" : "; " + String.join("; ", notes);
}
@Override
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupResetOffsetToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupResetOffsetToolHandlerTest.java
index b3d9cdd00..816813500 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupResetOffsetToolHandlerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupResetOffsetToolHandlerTest.java
@@ -6,13 +6,17 @@
*/
package org.apache.rocketmq.studio.ops.ai.tool.handler.group;
+import java.util.List;
import java.util.Map;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
+import
org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetQueuePreviewVO;
import org.apache.rocketmq.studio.instance.topic.MetadataService;
import org.apache.rocketmq.studio.ops.ai.tool.core.ToolExecutionContext;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.group.GroupResetOffsetInput;
import org.apache.rocketmq.studio.ops.ai.tool.contract.group.ResetOffsetOutput;
+import org.apache.rocketmq.studio.ops.ai.tool.contract.plan.ToolPlan;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
@@ -26,6 +30,7 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class GroupResetOffsetToolHandlerTest {
@@ -100,4 +105,61 @@ class GroupResetOffsetToolHandlerTest {
.satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
verifyNoInteractions(metadataService);
}
+
+ @Test
+ void previewShouldRenderAnUnavailableLagInsteadOfTheSentinelTest() {
+ // An offline group has no current offset, so the preview reports the
-1 "cannot be
+ // determined" sentinel for the current lag. The plan prose must never
interpolate it, or
+ // the operator is told "projected lag changes from -1 to 500".
+ GroupResetOffsetInput input = new GroupResetOffsetInput(
+ "untrusted-instance", "group-1", "TopicA", 1700000000000L);
+ ToolExecutionContext execution = ToolExecutionContext.of(
+ "instance-a", null, Map.of("instanceId",
"untrusted-instance"));
+ ResetConsumerOffsetPreviewVO preview =
ResetConsumerOffsetPreviewVO.builder()
+ .instanceId("instance-a")
+ .groupName("group-1")
+ .topic("TopicA")
+ .timestamp(1700000000000L)
+ .complete(true)
+ .allowReset(true)
+ .queueCount(1)
+ .warningCount(1)
+ .currentTotalLag(-1L)
+ .projectedTotalLag(500L)
+ .totalOffsetDelta(0L)
+ .warnings(List.of(
+ "Consumer group is not online and no consume offset
data is available"
+ + "; current offsets are unavailable, so
offset deltas cannot be computed"))
+ .queues(List.of(ResetConsumerOffsetQueuePreviewVO.builder()
+ .topic("TopicA")
+ .broker("broker-a")
+ .queueId(0)
+ .minOffset(10L)
+ .maxOffset(600L)
+ .brokerOffset(600L)
+ .consumerOffset(-1L)
+ .targetOffset(100L)
+ .currentLag(-1L)
+ .projectedLag(500L)
+ .offsetDelta(0L)
+ .riskLevel("INFO")
+ .message("Current offset unavailable (group offline);
target offset 100")
+ .build()))
+ .build();
+ when(metadataService.previewResetOffset("instance-a", "group-1",
1700000000000L, "TopicA"))
+ .thenReturn(preview);
+
+ ToolPlan plan = handler.preview(input, execution);
+
+ assertThat(plan.impact()).containsExactly(
+ "Moves offsets for 1 queues; projected lag changes from
unavailable to 500.");
+ assertThat(plan.impact()).noneMatch(line -> line.contains("-1"));
+ assertThat(plan.warnings()).containsExactly(
+ "Consumer group is not online and no consume offset data is
available"
+ + "; current offsets are unavailable, so offset deltas
cannot be computed");
+ // The structured before/after state keeps the documented sentinel
rather than a fabricated
+ // zero, so the caller can still tell "unknown" from "measured as
nothing".
+ assertThat(((Number)
plan.before().get("totalLag")).longValue()).isEqualTo(-1L);
+ assertThat(((Number)
plan.after().get("totalLag")).longValue()).isEqualTo(500L);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
index dbe9c512a..79476ca96 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
@@ -101,6 +101,7 @@ import
org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
@@ -531,6 +532,116 @@ class RocketMQAdminClientImplTest {
anyLong(), anyBoolean());
}
+ @Test
+ void previewResetOffsetAllowsOfflineGroupFromTopicRouteTest() throws
Exception {
+ // An offline group has no consume stats, but resetOffsetNew still
applies offline: the
+ // preview must enumerate the topic route and allow the reset instead
of blocking it.
+ long timestamp = 1784246400000L;
+ when(adminExt.examineConsumeStats("cg-orders"))
+ .thenThrow(new
MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
+ "Not found the consumer group connection"));
+ QueueData queueData = new QueueData();
+ queueData.setBrokerName("broker-1");
+ queueData.setReadQueueNums(2);
+ queueData.setPerm(6);
+ TopicRouteData routeData = new TopicRouteData();
+ routeData.setQueueDatas(List.of(queueData));
+ when(adminExt.examineTopicRouteInfo("orders")).thenReturn(routeData);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
+ when(adminExt.minOffset(any(MessageQueue.class))).thenReturn(10L);
+ when(adminExt.maxOffset(any(MessageQueue.class))).thenReturn(100L);
+ when(adminExt.searchOffset(eq("10.0.0.1:10911"), eq("orders"),
anyInt(), eq(timestamp), anyLong()))
+ .thenReturn(50L);
+
+ ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
+ "instance-a", "cg-orders", timestamp, "orders");
+
+ assertThat(preview.isAllowReset()).isTrue();
+ assertThat(preview.isComplete()).isTrue();
+ assertThat(preview.getQueueCount()).isEqualTo(2);
+
assertThat(preview.getCurrentTotalLag()).isEqualTo(ConsumerLagResolver.UNKNOWN);
+ assertThat(preview.getProjectedTotalLag()).isEqualTo(100L);
+ assertThat(preview.getWarnings()).anyMatch(w -> w.contains("not
online"));
+ assertThat(preview.getQueues())
+ .extracting(ResetConsumerOffsetQueuePreviewVO::getQueueId,
+ ResetConsumerOffsetQueuePreviewVO::getTargetOffset,
+ ResetConsumerOffsetQueuePreviewVO::getCurrentLag)
+ .containsExactly(
+ org.assertj.core.groups.Tuple.tuple(0, 50L,
ConsumerLagResolver.UNKNOWN),
+ org.assertj.core.groups.Tuple.tuple(1, 50L,
ConsumerLagResolver.UNKNOWN));
+ }
+
+ @Test
+ void
previewResetOffsetReportsUnavailableCurrentOffsetForOfflineGroupTest() throws
Exception {
+ // The offline preview must not splice the delta-based online message
into its own: with no
+ // current offset there is no delta, and borrowing that clause read as
the broken
+ // "the reset moves the offset to Offset unchanged".
+ long timestamp = 1784246400000L;
+ when(adminExt.examineConsumeStats("cg-orders"))
+ .thenThrow(new
MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
+ "Not found the consumer group connection"));
+ QueueData queueData = new QueueData();
+ queueData.setBrokerName("broker-1");
+ queueData.setReadQueueNums(1);
+ queueData.setPerm(6);
+ TopicRouteData routeData = new TopicRouteData();
+ routeData.setQueueDatas(List.of(queueData));
+ when(adminExt.examineTopicRouteInfo("orders")).thenReturn(routeData);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
+ when(adminExt.minOffset(any(MessageQueue.class))).thenReturn(10L);
+ when(adminExt.maxOffset(any(MessageQueue.class))).thenReturn(50L);
+ when(adminExt.searchOffset(eq("10.0.0.1:10911"), eq("orders"),
anyInt(), eq(timestamp), anyLong()))
+ .thenReturn(900L);
+
+ ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
+ "instance-a", "cg-orders", timestamp, "orders");
+
+ // searchOffset answered 900, beyond the retained range, so the target
is clamped to the
+ // broker maximum and the message names which end of the range it
landed on.
+ assertThat(preview.getQueues()).singleElement().satisfies(row -> {
+ assertThat(row.getTargetOffset()).isEqualTo(50L);
+
assertThat(row.getConsumerOffset()).isEqualTo(ConsumerLagResolver.UNKNOWN);
+
assertThat(row.getCurrentLag()).isEqualTo(ConsumerLagResolver.UNKNOWN);
+ assertThat(row.getProjectedLag()).isZero();
+ assertThat(row.getMessage()).isEqualTo(
+ "Current offset unavailable (group offline); target offset
50; "
+ + "target is the latest offset");
+ assertThat(row.getMessage()).doesNotContain("Offset unchanged");
+ });
+ assertThat(preview.getWarnings()).anySatisfy(warning -> {
+ assertThat(warning).contains("not online");
+ assertThat(warning).contains("current offsets are unavailable");
+ assertThat(warning).doesNotContain("-1");
+ });
+ }
+
+ @Test
+ void previewResetOffsetSkipsUnreadableQueuesForOfflineGroupTest() throws
Exception {
+ // resetOffsetByTimestampOld walks the read queues of the route; a
write-only queue is never
+ // consumed from, so the preview must not invent a target the reset
will not commit.
+ long timestamp = 1784246400000L;
+ when(adminExt.examineConsumeStats("cg-orders"))
+ .thenThrow(new
MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
+ "Not found the consumer group connection"));
+ QueueData writeOnly = new QueueData();
+ writeOnly.setBrokerName("broker-1");
+ writeOnly.setReadQueueNums(2);
+ writeOnly.setPerm(2);
+ TopicRouteData routeData = new TopicRouteData();
+ routeData.setQueueDatas(List.of(writeOnly));
+ when(adminExt.examineTopicRouteInfo("orders")).thenReturn(routeData);
+
+ ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
+ "instance-a", "cg-orders", timestamp, "orders");
+
+ assertThat(preview.isAllowReset()).isFalse();
+ assertThat(preview.getQueueCount()).isZero();
+ assertThat(preview.getWarnings())
+ .containsExactly("Consumer group is not online and no consume
offset data is available"
+ + "; no readable queues found for topic orders");
+ verify(adminExt, never()).searchOffset(anyString(), anyString(),
anyInt(), anyLong(), anyLong());
+ }
+
@Test
void previewResetOffsetShouldPreserveUnknownCurrentLagTest() throws
Exception {
long timestamp = 1784246400000L;
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index f244ec420..3ef910a1e 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -2587,6 +2587,15 @@ const translations: Record<string, Record<Lang, string>>
= {
'consumer.viewDistribution': { zh: '查看分布', en: 'View Distribution' },
'consumer.resetToTime': { zh: '重置到指定时间', en: 'Reset to Time' },
'consumer.resetSuccess': { zh: '位点重置成功', en: 'Offset reset successfully' },
+ 'consumer.resetPreview.previewFailed': { zh: '预览失败', en: 'Preview failed' },
+ 'consumer.resetPreview.targetOffsetUnavailable': {
+ zh: '目标位点不可用',
+ en: 'Target offset unavailable',
+ },
+ 'consumer.resetPreview.currentOffsetUnavailable': {
+ zh: '消费组离线，当前位点不可用；重置将按目标时间定位',
+ en: 'Group offline, current offset unavailable; the reset targets the
searched offset',
+ },
'consumer.fetchListFailed': {
zh: '消费组列表加载失败，请稍后重试',
en: 'Failed to load consumer groups. Please try again later.',
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 365c28a70..038661ddf 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -1228,6 +1228,90 @@ describe('Consumer page', () => {
expect(consumerService.resetConsumerOffset).not.toHaveBeenCalled();
});
+ it('previews an offline group without leaking the unknown sentinel', async
() => {
+ vi.mocked(consumerService.previewConsumerOffsetReset).mockResolvedValue({
+ instanceId: 'instance-1',
+ groupName: 'remote-cg',
+ topic: 'remote-topic',
+ timestamp: 1784246400000,
+ complete: true,
+ allowReset: true,
+ queueCount: 1,
+ warningCount: 2,
+ rewindQueueCount: 0,
+ fastForwardQueueCount: 0,
+ currentTotalLag: -1,
+ projectedTotalLag: 50,
+ totalOffsetDelta: 0,
+ warnings: [
+ 'At least one queue has unavailable lag; affected backlog totals are
unavailable',
+ 'Consumer group is not online and no consume offset data is available;
' +
+ 'current offsets are unavailable, so offset deltas cannot be
computed',
+ ],
+ queues: [
+ {
+ topic: 'remote-topic',
+ broker: 'broker-a',
+ queueId: 0,
+ minOffset: 10,
+ maxOffset: 100,
+ brokerOffset: 100,
+ // The group is offline: the broker still reports the target, but no
consumer reported a
+ // current offset, so the backend sends the -1 sentinel for the
current offset and lag.
+ consumerOffset: -1,
+ targetOffset: 50,
+ currentLag: -1,
+ projectedLag: 50,
+ offsetDelta: 0,
+ riskLevel: 'INFO',
+ message: 'Current offset unavailable (group offline); target offset
50',
+ },
+ ],
+ });
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ renderWithProviders(<ConsumerPage />);
+
+ await user.click(await screen.findByRole('button', { name: /重置位点/ }));
+ await user.click(screen.getByRole('combobox', { name: '目标 Topic' }));
+ const option = await waitFor(() => {
+ const element = screen
+ .getAllByText('remote-topic')
+ .find((candidate) =>
candidate.classList.contains('ant-select-item-option-content'));
+ if (!element) throw new Error('Missing target Topic option');
+ return element;
+ });
+ await user.click(option);
+ await user.click(screen.getByRole('button', { name: /预览影响/ }));
+
+ // The 说明 column speaks the page's own language; the English server
message is built around a
+ // delta the offline preview cannot compute, so it must not be echoed here.
+ expect(
+ await screen.findByText('消费组离线，当前位点不可用；重置将按目标时间定位'),
+ ).toBeInTheDocument();
+ expect(
+ screen.queryByText('Current offset unavailable (group offline); target
offset 50'),
+ ).not.toBeInTheDocument();
+
+ const dialog = screen.getByRole('dialog');
+ const row = within(dialog)
+ .getAllByRole('row')
+ .find((candidate) => within(candidate).queryByText('broker-a'));
+ expect(row).toBeDefined();
+ const cells = within(row!)
+ .getAllByRole('cell')
+ .map((cell) => cell.textContent?.trim());
+ // 当前位点 / 变化 / 当前堆积 render the sentinel as '-', the way the consumer
progress table does.
+ expect(cells.filter((cell) => cell === '-')).toHaveLength(3);
+ expect(cells.join('|')).not.toContain('-1');
+ // The aggregate net delta is not a measurement either, so the summary
must not show 0.
+ const netChangeLabel = within(dialog).getByText('位点净变化');
+ const netChange = (netChangeLabel.closest('th') ??
netChangeLabel).nextElementSibling;
+ expect(netChange?.textContent?.trim()).toBe('-');
+
+ await waitFor(() => expect(screen.getByRole('button', { name: '确认重置'
})).toBeEnabled());
+ expect(consumerService.resetConsumerOffset).not.toHaveBeenCalled();
+ });
+
it('reloads same-named group diagnostics after changing the selected
instance', async () => {
vi.mocked(instanceService.listInstances).mockResolvedValue([
{
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index 92b59ba5b..fbc314ece 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -194,15 +194,23 @@ const issueSeverityLabel = (severity:
ConsumerGroupHealthIssue['severity']) => {
return '提示';
};
-const resetPreviewQueueMessage = (queue: ResetConsumerOffsetQueuePreview) => {
+const resetPreviewQueueMessage = (
+ queue: ResetConsumerOffsetQueuePreview,
+ t: (key: string) => string,
+) => {
const messages: string[] = [];
if (queue.riskLevel === 'ERROR') {
- return queue.message || '预览失败';
+ return queue.message || t('consumer.resetPreview.previewFailed');
}
- if (queue.targetOffset < 0 || queue.consumerOffset < 0) {
- return queue.message || '目标位点不可用';
+ if (queue.targetOffset < 0) {
+ return t('consumer.resetPreview.targetOffsetUnavailable');
}
- if (queue.offsetDelta < 0) {
+ if (queue.consumerOffset < 0) {
+ // The group is offline: the broker still reports the target, but no
consumer reported a
+ // current offset, so the delta is not a measurement. Never fall back to
queue.message here —
+ // it is English prose built around a delta and would read as a broken
sentence in this column.
+ messages.push(t('consumer.resetPreview.currentOffsetUnavailable'));
+ } else if (queue.offsetDelta < 0) {
messages.push(`将回放 ${Math.abs(queue.offsetDelta).toLocaleString()} 条消息`);
} else if (queue.offsetDelta > 0) {
messages.push(`将跳过 ${queue.offsetDelta.toLocaleString()} 条未消费消息`);
@@ -218,6 +226,14 @@ const resetPreviewQueueMessage = (queue:
ResetConsumerOffsetQueuePreview) => {
return messages.join('；');
};
+/**
+ * True when any previewed queue has no current offset the broker could report
(an offline group),
+ * which makes the aggregate net delta a fabricated zero rather than a
measurement. The per-queue
+ * 变化 column already renders such rows as '-', so the summary must not claim a
number either.
+ */
+const hasUnavailableCurrentOffset = (queues:
ResetConsumerOffsetQueuePreview[]) =>
+ queues.some((queue) => queue.consumerOffset < 0 || queue.targetOffset < 0);
+
// Shared helper exported alongside the page component; fast-refresh rule
waived.
// eslint-disable-next-line react-refresh/only-export-components
export const diagnosticCacheKey = (instanceId: string | undefined, groupName:
string) =>
@@ -1487,11 +1503,14 @@ const ConsumerPageContent = ({
key: 'message',
width: 240,
ellipsis: true,
- render: (_: unknown, record: ResetConsumerOffsetQueuePreview) => (
- <Text style={{ fontSize: 14 }}
title={resetPreviewQueueMessage(record)}>
- {resetPreviewQueueMessage(record)}
- </Text>
- ),
+ render: (_: unknown, record: ResetConsumerOffsetQueuePreview) => {
+ const message = resetPreviewQueueMessage(record, t);
+ return (
+ <Text style={{ fontSize: 14 }} title={message}>
+ {message}
+ </Text>
+ );
+ },
},
];
@@ -2811,7 +2830,9 @@ const ConsumerPageContent = ({
{formatOffsetValue(resetPreview.projectedTotalLag)}
</Descriptions.Item>
<Descriptions.Item label="位点净变化">
- {formatOffsetDelta(resetPreview.totalOffsetDelta)}
+ {hasUnavailableCurrentOffset(resetPreviewQueues)
+ ? '-'
+ : formatOffsetDelta(resetPreview.totalOffsetDelta)}
</Descriptions.Item>
<Descriptions.Item label="回放 Queue">
{resetPreview.rewindQueueCount}
