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 d3469e2c3 feat(group): preview consumer offset reset impact before
applying (#2816)
d3469e2c3 is described below
commit d3469e2c3b9c26eb30538af0d48a7a722785c8fc
Author: coder999o <[email protected]>
AuthorDate: Mon Aug 31 21:15:17 2026 +0800
feat(group): preview consumer offset reset impact before applying (#2816)
---
.../instance/group/ConsumerGroupController.java | 7 +
.../group/ResetConsumerOffsetPreviewVO.java | 46 +++
.../group/ResetConsumerOffsetQueuePreviewVO.java | 42 ++
.../studio/instance/topic/MetadataService.java | 9 +
.../rocketmq/studio/provider/InstanceProvider.java | 48 +++
.../studio/provider/apache/AdminClient.java | 2 +
.../provider/apache/ApacheInstanceProvider.java | 7 +
.../provider/apache/RocketMQAdminClientImpl.java | 214 +++++++++++
.../group/ConsumerGroupControllerTest.java | 59 +++
.../studio/instance/topic/MetadataServiceTest.java | 38 ++
.../alibaba/AliyunInstanceProviderTest.java | 32 ++
.../apache/RocketMQAdminClientImplTest.java | 115 ++++++
.../tencent/TencentInstanceProviderTest.java | 24 ++
web/src/api/consumerGroups.test.ts | 49 +++
web/src/api/metadata.ts | 42 ++
.../pages/instance/__tests__/ConsumerPage.test.tsx | 155 +++++++-
web/src/pages/instance/consumer.tsx | 427 ++++++++++++++++++---
web/src/services/consumerService.test.ts | 63 +++
web/src/services/consumerService.ts | 100 +++++
19 files changed, 1421 insertions(+), 58 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
index 8c965bede..99d727c27 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
@@ -157,4 +157,11 @@ public class ConsumerGroupController {
.distinct()
.toList();
}
+
+ @PostMapping("/reset-offset/preview")
+ public Result<ResetConsumerOffsetPreviewVO> previewResetOffset(@Valid
@RequestBody ResetConsumerOffsetDTO request) {
+ return
Result.ok(metadataService.previewResetOffset(request.getInstanceId(),
request.getName(),
+ request.getTimestamp(), request.getTopic()));
+
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetPreviewVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetPreviewVO.java
new file mode 100644
index 000000000..5533b09bb
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetPreviewVO.java
@@ -0,0 +1,46 @@
+/*
+ * 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.studio.instance.group;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ResetConsumerOffsetPreviewVO {
+ private String instanceId;
+ private String groupName;
+ private String topic;
+ private long timestamp;
+ private boolean complete;
+ private boolean allowReset;
+ private int queueCount;
+ private int warningCount;
+ private int rewindQueueCount;
+ private int fastForwardQueueCount;
+ private long currentTotalLag;
+ private long projectedTotalLag;
+ private long totalOffsetDelta;
+ private List<String> warnings;
+ private List<ResetConsumerOffsetQueuePreviewVO> queues;
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetQueuePreviewVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetQueuePreviewVO.java
new file mode 100644
index 000000000..e37ae26c3
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetQueuePreviewVO.java
@@ -0,0 +1,42 @@
+/*
+ * 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.studio.instance.group;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ResetConsumerOffsetQueuePreviewVO {
+ private String topic;
+ private String broker;
+ private int queueId;
+ private long minOffset;
+ private long maxOffset;
+ private long brokerOffset;
+ private long consumerOffset;
+ private long targetOffset;
+ private long currentLag;
+ private long projectedLag;
+ private long offsetDelta;
+ private String riskLevel;
+ private String message;
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
index f482401a4..3dee45338 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
@@ -34,6 +34,7 @@ import org.springframework.util.StringUtils;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupSettingsVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
import org.apache.rocketmq.studio.provider.InstanceProvider;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
@@ -317,6 +318,14 @@ public class MetadataService {
resetOffset(null, name, timestamp, topic);
}
+ public ResetConsumerOffsetPreviewVO previewResetOffset(String instanceId,
String name,
+ long timestamp,
String topic) {
+ instanceId = normalizeInstanceId(instanceId);
+ String groupName = requireName(name, "consumer group name");
+ String topicName = requireName(topic, "topic name");
+ return resolve(instanceId).previewResetOffset(instanceId, groupName,
timestamp, topicName);
+ }
+
public void resetOffset(String instanceId, String name, long timestamp,
String topic) {
instanceId = normalizeInstanceId(instanceId);
String groupName = requireName(name, "consumer group name");
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
index e67849761..f12d3fcf7 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
@@ -22,6 +22,8 @@ import
org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.util.Pagination;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
+import
org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetQueuePreviewVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
import org.apache.rocketmq.studio.instance.message.DirectConsumeMessageDTO;
@@ -33,6 +35,7 @@ import org.apache.rocketmq.studio.instance.topic.TopicVO;
import java.util.List;
import java.util.Set;
+import java.util.ArrayList;
/**
* Unified instance-scoped operations SPI. Every method takes the Studio
instance id as its
@@ -105,6 +108,51 @@ public interface InstanceProvider {
List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId, String
groupName);
+ default ResetConsumerOffsetPreviewVO previewResetOffset(String instanceId,
String groupName,
+ long timestamp,
String topic) {
+ List<QueueProgressVO> progressRows = getGroupProgress(instanceId,
groupName);
+ List<ResetConsumerOffsetQueuePreviewVO> queues = new ArrayList<>();
+ for (QueueProgressVO progress : progressRows) {
+ if (progress == null || !topic.equals(progress.getTopic())) {
+ continue;
+ }
+ queues.add(ResetConsumerOffsetQueuePreviewVO.builder()
+ .topic(progress.getTopic())
+ .broker(progress.getBroker())
+ .queueId(progress.getQueueId())
+ .minOffset(-1L)
+ .maxOffset(-1L)
+ .brokerOffset(progress.getBrokerOffset())
+ .consumerOffset(progress.getConsumerOffset())
+ .targetOffset(-1L)
+ .currentLag(progress.getDiffTotal())
+ .projectedLag(-1L)
+ .offsetDelta(0L)
+ .riskLevel("WARNING")
+ .message("Provider does not expose target offset preview;
current lag is shown before reset")
+ .build());
+ }
+ List<String> warnings = List.of(
+ "Provider does not expose per-queue target offset preview;
confirm with current lag only");
+ return ResetConsumerOffsetPreviewVO.builder()
+ .instanceId(instanceId)
+ .groupName(groupName)
+ .topic(topic)
+ .timestamp(timestamp)
+ .complete(false)
+ .allowReset(!queues.isEmpty())
+ .queueCount(queues.size())
+ .warningCount(warnings.size())
+ .rewindQueueCount(0)
+ .fastForwardQueueCount(0)
+
.currentTotalLag(queues.stream().mapToLong(ResetConsumerOffsetQueuePreviewVO::getCurrentLag).sum())
+ .projectedTotalLag(-1L)
+ .totalOffsetDelta(0L)
+ .warnings(warnings)
+ .queues(queues)
+ .build();
+ }
+
void resetOffset(String instanceId, String groupName, long timestamp,
String topic);
List<MessageRecordVO> queryMessages(String instanceId, String topic,
String msgId,
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
index 36ecea467..1c4f91c29 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
@@ -21,6 +21,7 @@ import
org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
import org.apache.rocketmq.studio.instance.topic.TopicVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupSettingsVO;
+import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
public interface AdminClient {
@@ -35,5 +36,6 @@ public interface AdminClient {
ConsumerGroupSettingsVO updateConsumerGroupSettings(String instanceId,
String name, int retryQueueNums,
int retryMaxTimes);
void deleteConsumerGroup(String instanceId, String name);
+ ResetConsumerOffsetPreviewVO previewResetOffset(String instanceId, String
name, long timestamp, String topic);
void resetOffset(String instanceId, String name, long timestamp, String
topic);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
index 49c32786a..d80d1fba2 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
@@ -21,6 +21,7 @@ import
org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.instance.InstanceRepository;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
import org.apache.rocketmq.studio.instance.message.MessageProvider;
import org.apache.rocketmq.studio.instance.message.DirectConsumeMessageDTO;
@@ -148,6 +149,12 @@ public class ApacheInstanceProvider implements
InstanceProvider {
return metadataProvider.getGroupSubscriptions(instanceId, groupName);
}
+ @Override
+ public ResetConsumerOffsetPreviewVO previewResetOffset(String instanceId,
String groupName,
+ long timestamp,
String topic) {
+ return adminClient.previewResetOffset(instanceId, groupName,
timestamp, topic);
+ }
+
@Override
public void resetOffset(String instanceId, String groupName, long
timestamp, String topic) {
adminClient.resetOffset(instanceId, groupName, timestamp, topic);
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 db403c947..cd1ae9aa3 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.TopicAttributes;
import org.apache.rocketmq.common.message.Message;
+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;
@@ -38,6 +39,8 @@ import
org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupSettingsVO;
+import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
+import
org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetQueuePreviewVO;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
import org.apache.rocketmq.studio.instance.topic.TopicVO;
@@ -57,7 +60,9 @@ import lombok.extern.slf4j.Slf4j;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -73,6 +78,10 @@ public class RocketMQAdminClientImpl implements AdminClient {
private static final int MAX_MESSAGE_SIZE = 4 * 1024 * 1024; // 4 MB
default broker limit
private static final String LEGACY_METADATA_SCOPE = "";
+ private static final long RESET_OFFSET_PREVIEW_TIMEOUT_MILLIS = 3_000L;
+ private static final String RISK_INFO = "INFO";
+ private static final String RISK_WARNING = "WARNING";
+ private static final String RISK_ERROR = "ERROR";
private final MqAdminExtFactory adminFactory;
private final RocketMQProperties properties;
@@ -660,6 +669,211 @@ public class RocketMQAdminClientImpl implements
AdminClient {
}
}
+ @Override
+ public ResetConsumerOffsetPreviewVO previewResetOffset(String instanceId,
String name,
+ long timestamp,
String topic) {
+ if (!StringUtils.hasText(topic)) {
+ throw new BusinessException(400, "topic is required for offset
reset preview");
+ }
+ String topicName = topic.trim();
+ try {
+ return executeForInstance(instanceId, admin ->
doPreviewResetOffset(
+ instanceId, admin, name, timestamp, topicName));
+ } catch (BusinessException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new BusinessException(500, "Failed to preview reset offset:
" + e.getMessage());
+ }
+ }
+
+ private ResetConsumerOffsetPreviewVO doPreviewResetOffset(String
instanceId, MQAdminExt admin,
+ String name,
long timestamp, String topic) {
+ try {
+ ConsumeStats stats = admin.examineConsumeStats(name);
+ if (stats == null || stats.getOffsetTable() == null ||
stats.getOffsetTable().isEmpty()) {
+ return emptyResetOffsetPreview(instanceId, name, timestamp,
topic,
+ "No consume offset data found for consumer group " +
name);
+ }
+
+ List<ResetConsumerOffsetQueuePreviewVO> queues = new ArrayList<>();
+ for (Map.Entry<MessageQueue, OffsetWrapper> entry :
stats.getOffsetTable().entrySet()) {
+ MessageQueue queue = entry.getKey();
+ if (queue == null || !topic.equals(queue.getTopic())) {
+ continue;
+ }
+ queues.add(previewResetOffsetQueue(admin, queue,
entry.getValue(), timestamp));
+ }
+ queues.sort(Comparator
+ .comparing((ResetConsumerOffsetQueuePreviewVO queue) ->
+ queue.getBroker() == null ? "" :
queue.getBroker(),
+ String.CASE_INSENSITIVE_ORDER)
+
.thenComparingInt(ResetConsumerOffsetQueuePreviewVO::getQueueId));
+ if (queues.isEmpty()) {
+ return emptyResetOffsetPreview(instanceId, name, timestamp,
topic,
+ "No consume offset data found for topic " + topic);
+ }
+
+ long currentTotalLag =
queues.stream().mapToLong(ResetConsumerOffsetQueuePreviewVO::getCurrentLag).sum();
+ long projectedTotalLag = queues.stream()
+
.mapToLong(ResetConsumerOffsetQueuePreviewVO::getProjectedLag)
+ .sum();
+ long totalOffsetDelta =
queues.stream().mapToLong(ResetConsumerOffsetQueuePreviewVO::getOffsetDelta).sum();
+ int rewindQueueCount = (int) queues.stream().filter(queue ->
queue.getOffsetDelta() < 0).count();
+ int fastForwardQueueCount = (int) queues.stream().filter(queue ->
queue.getOffsetDelta() > 0).count();
+ List<String> warnings = buildResetOffsetPreviewWarnings(queues,
rewindQueueCount, fastForwardQueueCount);
+ 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(rewindQueueCount)
+ .fastForwardQueueCount(fastForwardQueueCount)
+ .currentTotalLag(currentTotalLag)
+ .projectedTotalLag(projectedTotalLag)
+ .totalOffsetDelta(totalOffsetDelta)
+ .warnings(warnings)
+ .queues(queues)
+ .build();
+ } catch (Exception e) {
+ if (isConsumerNotOnline(e)) {
+ return emptyResetOffsetPreview(instanceId, name, timestamp,
topic,
+ "Consumer group is not online and no consume offset
data is available");
+ }
+ throw new BusinessException(500, "Failed to preview reset offset:
" + e.getMessage());
+ }
+ }
+
+ private ResetConsumerOffsetQueuePreviewVO
previewResetOffsetQueue(MQAdminExt admin, MessageQueue queue,
+
OffsetWrapper wrapper, long timestamp) {
+ long brokerOffset = wrapper == null ? 0L : wrapper.getBrokerOffset();
+ long consumerOffset = wrapper == null ? 0L :
wrapper.getConsumerOffset();
+ long currentLag = resolveLag(brokerOffset, consumerOffset);
+ try {
+ long minOffset = admin.minOffset(queue);
+ long maxOffset = admin.maxOffset(queue);
+ long targetOffset = admin.searchOffset(queue.getBrokerName(),
queue.getTopic(), queue.getQueueId(),
+ timestamp, RESET_OFFSET_PREVIEW_TIMEOUT_MILLIS);
+ targetOffset = clampOffset(targetOffset, minOffset, maxOffset);
+ long offsetDelta = targetOffset - consumerOffset;
+ long projectedLag = resolveLag(brokerOffset, targetOffset);
+ return ResetConsumerOffsetQueuePreviewVO.builder()
+ .topic(queue.getTopic())
+ .broker(queue.getBrokerName())
+ .queueId(queue.getQueueId())
+ .minOffset(minOffset)
+ .maxOffset(maxOffset)
+ .brokerOffset(brokerOffset)
+ .consumerOffset(consumerOffset)
+ .targetOffset(targetOffset)
+ .currentLag(currentLag)
+ .projectedLag(projectedLag)
+ .offsetDelta(offsetDelta)
+ .riskLevel(resetOffsetRiskLevel(offsetDelta, targetOffset,
minOffset, maxOffset))
+ .message(resetOffsetPreviewMessage(offsetDelta,
targetOffset, minOffset, maxOffset))
+ .build();
+ } catch (Exception e) {
+ return ResetConsumerOffsetQueuePreviewVO.builder()
+ .topic(queue.getTopic())
+ .broker(queue.getBrokerName())
+ .queueId(queue.getQueueId())
+ .minOffset(-1L)
+ .maxOffset(-1L)
+ .brokerOffset(brokerOffset)
+ .consumerOffset(consumerOffset)
+ .targetOffset(consumerOffset)
+ .currentLag(currentLag)
+ .projectedLag(currentLag)
+ .offsetDelta(0L)
+ .riskLevel(RISK_ERROR)
+ .message("Failed to preview queue offset: " +
e.getMessage())
+ .build();
+ }
+ }
+
+ private ResetConsumerOffsetPreviewVO emptyResetOffsetPreview(String
instanceId, String name,
+ long
timestamp, String topic, String warning) {
+ return ResetConsumerOffsetPreviewVO.builder()
+ .instanceId(instanceId)
+ .groupName(name)
+ .topic(topic)
+ .timestamp(timestamp)
+ .complete(false)
+ .allowReset(false)
+ .queueCount(0)
+ .warningCount(1)
+ .rewindQueueCount(0)
+ .fastForwardQueueCount(0)
+ .currentTotalLag(0)
+ .projectedTotalLag(0)
+ .totalOffsetDelta(0)
+ .warnings(List.of(warning))
+ .queues(List.of())
+ .build();
+ }
+
+ private List<String>
buildResetOffsetPreviewWarnings(List<ResetConsumerOffsetQueuePreviewVO> queues,
+ int rewindQueueCount,
int fastForwardQueueCount) {
+ List<String> warnings = new ArrayList<>();
+ long failedQueueCount = queues.stream().filter(queue ->
RISK_ERROR.equals(queue.getRiskLevel())).count();
+ if (failedQueueCount > 0) {
+ warnings.add("Failed to preview " + failedQueueCount + " queue(s);
retry before applying the reset");
+ }
+ if (fastForwardQueueCount > 0) {
+ warnings.add(fastForwardQueueCount + " queue(s) will move forward
and may skip unconsumed messages");
+ }
+ if (rewindQueueCount > 0) {
+ warnings.add(rewindQueueCount + " queue(s) will move backward and
may replay consumed messages");
+ }
+ if (queues.stream().anyMatch(queue -> queue.getMinOffset() >= 0
+ && queue.getTargetOffset() == queue.getMinOffset())) {
+ warnings.add("At least one queue will reset to the minimum
retained offset");
+ }
+ if (queues.stream().anyMatch(queue -> queue.getMaxOffset() >= 0
+ && queue.getTargetOffset() == queue.getMaxOffset())) {
+ warnings.add("At least one queue will reset to the latest offset");
+ }
+ return warnings;
+ }
+
+ private long resolveLag(long brokerOffset, long consumerOffset) {
+ return Math.max(0L, brokerOffset - consumerOffset);
+ }
+
+ private long clampOffset(long offset, long minOffset, long maxOffset) {
+ return Math.max(minOffset, Math.min(offset, maxOffset));
+ }
+
+ private String resetOffsetRiskLevel(long offsetDelta, long targetOffset,
long minOffset, long maxOffset) {
+ if (offsetDelta != 0 || targetOffset == minOffset || targetOffset ==
maxOffset) {
+ return RISK_WARNING;
+ }
+ return RISK_INFO;
+ }
+
+ private String resetOffsetPreviewMessage(long offsetDelta, long
targetOffset, long minOffset, long maxOffset) {
+ List<String> messages = new ArrayList<>();
+ if (offsetDelta < 0) {
+ messages.add("Replays " + Math.abs(offsetDelta) + " message(s)");
+ } else if (offsetDelta > 0) {
+ messages.add("Skips " + offsetDelta + " unconsumed message(s)");
+ } else {
+ messages.add("Offset unchanged");
+ }
+ if (targetOffset == minOffset) {
+ messages.add("target is the minimum retained offset");
+ }
+ if (targetOffset == maxOffset) {
+ messages.add("target is the latest offset");
+ }
+ return String.join("; ", messages);
+ }
+
@Override
public void resetOffset(String instanceId, String name, long timestamp,
String topic) {
if (!StringUtils.hasText(topic)) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
index 04a23a7f7..736dd3136 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
@@ -319,6 +319,65 @@ class ConsumerGroupControllerTest {
verify(metadataService).resetOffset(eq("instance-a"), eq("cg-orders"),
eq(1784246400000L), eq("orders"));
}
+ @Test
+ void previewResetOffsetShouldPassValidatedRequestAndReturnQueueImpact()
throws Exception {
+ Map<String, Object> body = Map.of(
+ "instanceId", "instance-a",
+ "name", "cg-orders",
+ "topic", "orders",
+ "timestamp", 1784246400000L
+ );
+ ResetConsumerOffsetQueuePreviewVO queue =
ResetConsumerOffsetQueuePreviewVO.builder()
+ .topic("orders")
+ .broker("broker-a")
+ .queueId(0)
+ .minOffset(0L)
+ .maxOffset(200L)
+ .brokerOffset(120L)
+ .consumerOffset(90L)
+ .targetOffset(80L)
+ .currentLag(30L)
+ .projectedLag(40L)
+ .offsetDelta(-10L)
+ .riskLevel("WARNING")
+ .message("Replays 10 message(s)")
+ .build();
+ ResetConsumerOffsetPreviewVO preview =
ResetConsumerOffsetPreviewVO.builder()
+ .instanceId("instance-a")
+ .groupName("cg-orders")
+ .topic("orders")
+ .timestamp(1784246400000L)
+ .complete(true)
+ .allowReset(true)
+ .queueCount(1)
+ .warningCount(1)
+ .rewindQueueCount(1)
+ .fastForwardQueueCount(0)
+ .currentTotalLag(30L)
+ .projectedTotalLag(40L)
+ .totalOffsetDelta(-10L)
+ .warnings(List.of("1 queue(s) will move backward and may
replay consumed messages"))
+ .queues(List.of(queue))
+ .build();
+ when(metadataService.previewResetOffset("instance-a", "cg-orders",
1784246400000L, "orders"))
+ .thenReturn(preview);
+
+ mockMvc.perform(post("/api/groups/reset-offset/preview")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(body)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data.groupName").value("cg-orders"))
+ .andExpect(jsonPath("$.data.allowReset").value(true))
+ .andExpect(jsonPath("$.data.queueCount").value(1))
+ .andExpect(jsonPath("$.data.projectedTotalLag").value(40))
+ .andExpect(jsonPath("$.data.queues[0].targetOffset").value(80))
+
.andExpect(jsonPath("$.data.queues[0].riskLevel").value("WARNING"));
+
+ verify(metadataService).previewResetOffset(eq("instance-a"),
eq("cg-orders"),
+ eq(1784246400000L), eq("orders"));
+ }
+
@Test
void deleteConsumerGroupShouldReturnSuccess() throws Exception {
mockMvc.perform(post("/api/groups/delete")
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
index 82207638a..ab6205bce 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
@@ -28,6 +28,8 @@ import
org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.CreateConsumerGroupDTO;
import org.apache.rocketmq.studio.instance.group.ImportConsumerGroupsResultVO;
+import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
+
import org.apache.rocketmq.studio.provider.InstanceProvider;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.apache.rocketmq.studio.provider.apache.AdminClient;
@@ -609,6 +611,41 @@ class MetadataServiceTest {
verifyNoInteractions(apacheProvider);
}
+ @Test
+ void previewResetOffsetShouldNormalizeAndDelegateToProvider() {
+ ResetConsumerOffsetPreviewVO preview =
ResetConsumerOffsetPreviewVO.builder()
+ .instanceId("instance-a")
+ .groupName("cg-orders")
+ .topic("orders")
+ .timestamp(1784246400000L)
+ .complete(true)
+ .allowReset(true)
+ .queueCount(0)
+ .warnings(List.of())
+ .queues(List.of())
+ .build();
+ when(apacheProvider.previewResetOffset("instance-a", "cg-orders",
1784246400000L, "orders"))
+ .thenReturn(preview);
+
+ ResetConsumerOffsetPreviewVO result =
metadataService.previewResetOffset(
+ "instance-a", " cg-orders ", 1784246400000L, " orders ");
+
+ assertThat(result).isSameAs(preview);
+ verify(apacheProvider).previewResetOffset("instance-a", "cg-orders",
1784246400000L, "orders");
+ }
+
+ @Test
+ void previewResetOffsetShouldRejectBlankTopicBeforeProviderResolution() {
+ assertThatThrownBy(() ->
metadataService.previewResetOffset("instance-a", "cg-orders",
+ 1784246400000L, " "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("topic name is required")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+
+
+ verifyNoInteractions(apacheProvider);
+ }
+
private ConsumerGroupVO consumerGroup(String name, String namespace, long
lag, SubscriptionMode mode) {
ConsumerGroupVO group = new ConsumerGroupVO();
group.setName(name);
@@ -669,4 +706,5 @@ class MetadataServiceTest {
}
+
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
index 676ecf12b..5b43f55c6 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
@@ -44,6 +44,7 @@ import org.apache.rocketmq.studio.instance.InstanceRepository;
import org.apache.rocketmq.studio.instance.InstanceVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
import org.apache.rocketmq.studio.instance.message.TraceNodeVO;
import org.apache.rocketmq.studio.instance.message.TraceRecordVO;
@@ -258,6 +259,37 @@ class AliyunInstanceProviderTest {
assertThat(totalRow.getDiffTotal()).isEqualTo(100L);
}
+ @Test
+ void previewResetOffsetShouldAllowLimitedCloudPreviewTest() {
+ stubInstance();
+ stubCallThrough();
+ GetConsumerGroupLagResponse response =
GetConsumerGroupLagResponse.create().toBuilder()
+ .statusCode(200)
+ .body(GetConsumerGroupLagResponseBody.builder()
+ .data(GetConsumerGroupLagResponseBody.Data.builder()
+ .consumerGroupId("GID_test")
+ .topicLagMap(Map.of("topic-a",
+
DataTopicLagMapValue.builder().readyCount(42L).build()))
+ .build())
+ .build())
+ .build();
+ when(asyncClient.getConsumerGroupLag(any()))
+ .thenReturn(CompletableFuture.completedFuture(response));
+
+ ResetConsumerOffsetPreviewVO preview = provider.previewResetOffset(
+ STUDIO_INSTANCE_ID, "GID_test", 1679458628000L, "topic-a");
+
+ assertThat(preview.isComplete()).isFalse();
+ assertThat(preview.isAllowReset()).isTrue();
+ assertThat(preview.getQueueCount()).isEqualTo(1);
+ assertThat(preview.getCurrentTotalLag()).isEqualTo(42L);
+ assertThat(preview.getProjectedTotalLag()).isEqualTo(-1L);
+ assertThat(preview.getWarnings())
+ .containsExactly("Provider does not expose per-queue target
offset preview; confirm with current lag only");
+
assertThat(preview.getQueues().get(0).getTargetOffset()).isEqualTo(-1L);
+
assertThat(preview.getQueues().get(0).getRiskLevel()).isEqualTo("WARNING");
+ }
+
@Test
void queryMessagesShouldMapFieldsAndDecodeBase64BodyTest() {
stubInstance();
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 df80091c7..1ced0f60e 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
@@ -22,7 +22,10 @@ import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
@@ -35,6 +38,8 @@ import
org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupSettingsVO;
import org.apache.rocketmq.studio.instance.group.ConsumerInstanceVO;
+import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
+import
org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetQueuePreviewVO;
import org.apache.rocketmq.studio.instance.topic.TopicVO;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
@@ -62,6 +67,8 @@ 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.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
@@ -244,6 +251,107 @@ class RocketMQAdminClientImplTest {
"instanceId=instance-a, topic=orders,
timestamp=1784246400000", "SUCCESS");
}
+ @Test
+ void
previewResetOffsetShouldComputeQueueImpactWithoutMutatingBrokerOffsets() throws
Exception {
+ long timestamp = 1784246400000L;
+ ConsumeStats stats = new ConsumeStats();
+ MessageQueue rewindQueue = new MessageQueue("orders", "broker-a", 0);
+ MessageQueue fastForwardQueue = new MessageQueue("orders", "broker-b",
1);
+ MessageQueue skippedQueue = new MessageQueue("payments", "broker-a",
0);
+ stats.getOffsetTable().put(rewindQueue, offsetWrapper(120L, 90L));
+ stats.getOffsetTable().put(fastForwardQueue, offsetWrapper(200L,
150L));
+ stats.getOffsetTable().put(skippedQueue, offsetWrapper(50L, 40L));
+ when(adminExt.examineConsumeStats("cg-orders")).thenReturn(stats);
+ when(adminExt.minOffset(rewindQueue)).thenReturn(0L);
+ when(adminExt.maxOffset(rewindQueue)).thenReturn(200L);
+ when(adminExt.searchOffset("broker-a", "orders", 0, timestamp,
3_000L)).thenReturn(80L);
+ when(adminExt.minOffset(fastForwardQueue)).thenReturn(0L);
+ when(adminExt.maxOffset(fastForwardQueue)).thenReturn(220L);
+ when(adminExt.searchOffset("broker-b", "orders", 1, timestamp,
3_000L)).thenReturn(170L);
+
+ ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
+ null, "cg-orders", timestamp, "orders");
+
+ assertThat(preview.isComplete()).isTrue();
+ assertThat(preview.isAllowReset()).isTrue();
+ assertThat(preview.getQueueCount()).isEqualTo(2);
+ assertThat(preview.getCurrentTotalLag()).isEqualTo(80L);
+ assertThat(preview.getProjectedTotalLag()).isEqualTo(70L);
+ assertThat(preview.getRewindQueueCount()).isEqualTo(1);
+ assertThat(preview.getFastForwardQueueCount()).isEqualTo(1);
+ assertThat(preview.getTotalOffsetDelta()).isEqualTo(10L);
+ assertThat(preview.getWarnings())
+ .contains("1 queue(s) will move forward and may skip
unconsumed messages",
+ "1 queue(s) will move backward and may replay consumed
messages");
+ assertThat(preview.getQueues())
+ .extracting(ResetConsumerOffsetQueuePreviewVO::getBroker,
+ ResetConsumerOffsetQueuePreviewVO::getQueueId,
+ ResetConsumerOffsetQueuePreviewVO::getTargetOffset,
+ ResetConsumerOffsetQueuePreviewVO::getOffsetDelta,
+ ResetConsumerOffsetQueuePreviewVO::getProjectedLag)
+ .containsExactly(
+ org.assertj.core.groups.Tuple.tuple("broker-a", 0,
80L, -10L, 40L),
+ org.assertj.core.groups.Tuple.tuple("broker-b", 1,
170L, 20L, 30L));
+ verify(adminExt, never()).resetOffsetByTimestamp(anyString(),
anyString(), anyString(),
+ anyLong(), anyBoolean());
+ }
+
+ @Test
+ void previewResetOffsetShouldMarkFailedQueuesIncomplete() throws Exception
{
+ long timestamp = 1784246400000L;
+ ConsumeStats stats = new ConsumeStats();
+ MessageQueue queue = new MessageQueue("orders", "broker-a", 0);
+ stats.getOffsetTable().put(queue, offsetWrapper(120L, 90L));
+ when(adminExt.examineConsumeStats("cg-orders")).thenReturn(stats);
+ when(adminExt.minOffset(queue)).thenThrow(new
IllegalStateException("offset unavailable"));
+
+ ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
+ null, "cg-orders", timestamp, "orders");
+
+ assertThat(preview.isComplete()).isFalse();
+ assertThat(preview.isAllowReset()).isFalse();
+ assertThat(preview.getWarningCount()).isEqualTo(1);
+ assertThat(preview.getWarnings())
+ .containsExactly("Failed to preview 1 queue(s); retry before
applying the reset");
+ assertThat(preview.getQueues()).hasSize(1);
+ ResetConsumerOffsetQueuePreviewVO row = preview.getQueues().get(0);
+ assertThat(row.getRiskLevel()).isEqualTo("ERROR");
+ assertThat(row.getTargetOffset()).isEqualTo(90L);
+ assertThat(row.getProjectedLag()).isEqualTo(30L);
+ assertThat(row.getMessage()).contains("offset unavailable");
+ verify(adminExt, never()).resetOffsetByTimestamp(anyString(),
anyString(), anyString(),
+ anyLong(), anyBoolean());
+ }
+
+ @Test
+ void previewResetOffsetShouldReturnEmptyPreviewWhenTopicHasNoOffsets()
throws Exception {
+ ConsumeStats stats = new ConsumeStats();
+ stats.getOffsetTable().put(new MessageQueue("payments", "broker-a",
0), offsetWrapper(120L, 90L));
+ when(adminExt.examineConsumeStats("cg-orders")).thenReturn(stats);
+
+ ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
+ null, "cg-orders", 1784246400000L, "orders");
+
+ assertThat(preview.isComplete()).isFalse();
+ assertThat(preview.isAllowReset()).isFalse();
+ assertThat(preview.getQueueCount()).isZero();
+ assertThat(preview.getWarnings()).containsExactly("No consume offset
data found for topic orders");
+ verify(adminExt, never()).resetOffsetByTimestamp(anyString(),
anyString(), anyString(),
+ anyLong(), anyBoolean());
+ }
+
+ @Test
+ void previewResetOffsetShouldRejectBlankTopicBeforeResolvingAdmin() {
+ assertThatThrownBy(() -> adminClient.previewResetOffset("instance-a",
"cg-orders", 1784246400000L, " "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("topic is required for offset reset preview")
+ .satisfies(exception -> assertThat(((BusinessException)
exception).getCode()).isEqualTo(400));
+
+ verifyNoInteractions(runtimeAdminClientResolver);
+ verify(adminFactory, never()).execute(anyString(), any(), any());
+ verifyNoInteractions(auditService);
+ }
+
@Test
void resetOffsetShouldRejectBlankTopicBeforeResolvingAdmin() {
assertThatThrownBy(() -> adminClient.resetOffset("instance-a",
"cg-orders", 1784246400000L, " "))
@@ -733,6 +841,13 @@ class RocketMQAdminClientImplTest {
return clusterInfo;
}
+ private OffsetWrapper offsetWrapper(long brokerOffset, long
consumerOffset) {
+ OffsetWrapper wrapper = new OffsetWrapper();
+ wrapper.setBrokerOffset(brokerOffset);
+ wrapper.setConsumerOffset(consumerOffset);
+ return wrapper;
+ }
+
@Test
void sendMessageShouldNotFailWhenAuditRecordingFails() throws Exception {
doThrow(new RuntimeException("audit db down")).when(auditService)
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
index b93489163..234584860 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
@@ -51,6 +51,7 @@ import org.apache.rocketmq.studio.instance.InstanceRepository;
import org.apache.rocketmq.studio.instance.InstanceVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
import org.apache.rocketmq.studio.instance.message.TraceNodeVO;
@@ -590,6 +591,29 @@ class TencentInstanceProviderTest {
assertThat(subscriptions.get(0).getType()).isEqualTo("TAG");
}
+ @Test
+ void previewResetOffsetShouldAllowLimitedCloudPreviewTest() throws
Exception {
+ SubscriptionData subscription = new SubscriptionData();
+ subscription.setTopic("orders");
+ subscription.setConsumerLag(42L);
+ DescribeTopicListByGroupResponse response = new
DescribeTopicListByGroupResponse();
+ response.setData(new SubscriptionData[]{subscription});
+ when(client.DescribeTopicListByGroup(any())).thenReturn(response);
+
+ ResetConsumerOffsetPreviewVO preview = provider.previewResetOffset(
+ STUDIO_INSTANCE_ID, "GID_test", 1600000000000L, "orders");
+
+ assertThat(preview.isComplete()).isFalse();
+ assertThat(preview.isAllowReset()).isTrue();
+ assertThat(preview.getQueueCount()).isEqualTo(1);
+ assertThat(preview.getCurrentTotalLag()).isEqualTo(42L);
+ assertThat(preview.getProjectedTotalLag()).isEqualTo(-1L);
+ assertThat(preview.getWarnings())
+ .containsExactly("Provider does not expose per-queue target
offset preview; confirm with current lag only");
+
assertThat(preview.getQueues().get(0).getTargetOffset()).isEqualTo(-1L);
+
assertThat(preview.getQueues().get(0).getRiskLevel()).isEqualTo("WARNING");
+ }
+
@Test
void resetOffsetShouldCallTencentOpenApiTest() throws Exception {
when(client.ResetConsumerGroupOffset(any())).thenReturn(null);
diff --git a/web/src/api/consumerGroups.test.ts
b/web/src/api/consumerGroups.test.ts
index a25044ea1..66e28622a 100644
--- a/web/src/api/consumerGroups.test.ts
+++ b/web/src/api/consumerGroups.test.ts
@@ -26,6 +26,7 @@ import {
listConsumerGroupPage,
listConsumerGroups,
deleteConsumerGroup,
+ previewConsumerOffsetReset,
resetConsumerOffset,
updateConsumerGroupSettings,
} from './metadata';
@@ -118,6 +119,54 @@ describe('consumer groups API contract', () => {
await expect(resetConsumerOffset(reset)).resolves.toBeUndefined();
});
+ it('posts reset preview requests and unwraps queue impact data', async () =>
{
+ const request = {
+ name: group.name,
+ instanceId: 'instance-1',
+ topic: 'orders',
+ timestamp: 1784246400000,
+ };
+ const preview = {
+ instanceId: 'instance-1',
+ groupName: group.name,
+ topic: 'orders',
+ timestamp: 1784246400000,
+ complete: true,
+ allowReset: true,
+ queueCount: 1,
+ warningCount: 1,
+ rewindQueueCount: 1,
+ fastForwardQueueCount: 0,
+ currentTotalLag: 30,
+ projectedTotalLag: 40,
+ totalOffsetDelta: -10,
+ warnings: ['1 queue(s) will move backward and may replay consumed
messages'],
+ queues: [
+ {
+ topic: 'orders',
+ broker: 'broker-a',
+ queueId: 0,
+ minOffset: 0,
+ maxOffset: 200,
+ brokerOffset: 120,
+ consumerOffset: 90,
+ targetOffset: 80,
+ currentLag: 30,
+ projectedLag: 40,
+ offsetDelta: -10,
+ riskLevel: 'WARNING',
+ message: 'Replays 10 message(s)',
+ },
+ ],
+ };
+ mock.onPost('/groups/reset-offset/preview').reply((config) => {
+ expect(JSON.parse(config.data)).toEqual(request);
+ return [200, { code: 200, data: preview }];
+ });
+
+ await
expect(previewConsumerOffsetReset(request)).resolves.toEqual(preview);
+ });
+
it('includes selected instance context when deleting a consumer group',
async () => {
mock.onPost('/groups/delete').reply((config) => {
expect(JSON.parse(config.data)).toEqual({ name: group.name, instanceId:
'instance-1' });
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 148a8c842..185fe3271 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -373,6 +373,48 @@ export interface ResetConsumerOffsetRequest {
topic: string;
}
+export interface ResetConsumerOffsetQueuePreview {
+ topic: string;
+ broker: string;
+ queueId: number;
+ minOffset: number;
+ maxOffset: number;
+ brokerOffset: number;
+ consumerOffset: number;
+ targetOffset: number;
+ currentLag: number;
+ projectedLag: number;
+ offsetDelta: number;
+ riskLevel: 'INFO' | 'WARNING' | 'ERROR' | string;
+ message: string;
+}
+
+export interface ResetConsumerOffsetPreview {
+ instanceId?: string;
+ groupName: string;
+ topic: string;
+ timestamp: number;
+ complete: boolean;
+ allowReset: boolean;
+ queueCount: number;
+ warningCount: number;
+ rewindQueueCount: number;
+ fastForwardQueueCount: number;
+ currentTotalLag: number;
+ projectedTotalLag: number;
+ totalOffsetDelta: number;
+ warnings: string[];
+ queues: ResetConsumerOffsetQueuePreview[];
+}
+
+export async function previewConsumerOffsetReset(data:
ResetConsumerOffsetRequest) {
+ const res = await client.post<{ data: ResetConsumerOffsetPreview }>(
+ '/groups/reset-offset/preview',
+ data,
+ );
+ return res.data.data;
+}
+
export async function resetConsumerOffset(data: ResetConsumerOffsetRequest) {
await client.post('/groups/reset-offset', data);
}
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index eda784c6c..55ba19b2c 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -15,12 +15,12 @@
* limitations under the License.
*/
-import { App, Modal } from 'antd';
-import { act, fireEvent, render, screen, waitFor, within } from
'@testing-library/react';
+import { App, Modal, message } from 'antd';
+import { act, cleanup, fireEvent, render, screen, waitFor, within } from
'@testing-library/react';
import userEvent from '@testing-library/user-event';
import type React from 'react';
import { MemoryRouter } from 'react-router-dom';
-import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from
'vitest';
import type { ConsumerGroup } from '../../../api/metadata';
import * as instanceService from '../../../services/instanceService';
import { LangProvider } from '../../../i18n/LangContext';
@@ -39,6 +39,7 @@ vi.mock('../../../services/consumerService', () => ({
getConsumerGroupSettings: vi.fn(),
importConsumerGroups: vi.fn(),
listAllConsumerGroups: vi.fn(),
+ previewConsumerOffsetReset: vi.fn(),
updateConsumerGroupSettings: vi.fn(),
listConsumerGroupPage: vi.fn(),
refreshConsumerGroup: vi.fn(),
@@ -48,7 +49,7 @@ const instanceServiceMocks = vi.hoisted(() => ({
listInstances: vi.fn() }));
vi.mock('../../../services/instanceService', () => instanceServiceMocks);
-beforeAll(() => {
+const installBrowserMocks = () => {
Object.defineProperty(window, 'matchMedia', {
writable: true,
value: vi.fn().mockImplementation((query: string) => ({
@@ -70,7 +71,7 @@ beforeAll(() => {
writable: true,
value: vi.fn(),
});
-});
+};
const group: ConsumerGroup = {
name: 'remote-cg',
@@ -118,9 +119,13 @@ const renderWithProviders = (ui: React.ReactElement,
initialEntry = '/instance/c
</App>,
);
+beforeAll(installBrowserMocks);
+
describe('Consumer page', () => {
beforeEach(() => {
- vi.clearAllMocks();
+ vi.restoreAllMocks();
+ vi.resetAllMocks();
+ installBrowserMocks();
vi.mocked(instanceService.listInstances).mockResolvedValue([
{
id: 1,
@@ -205,6 +210,39 @@ describe('Consumer page', () => {
consistency: '一致',
},
]);
+ vi.mocked(consumerService.previewConsumerOffsetReset).mockResolvedValue({
+ instanceId: 'instance-1',
+ groupName: 'remote-cg',
+ topic: 'remote-topic',
+ timestamp: 1784246400000,
+ complete: true,
+ allowReset: true,
+ queueCount: 1,
+ warningCount: 1,
+ rewindQueueCount: 1,
+ fastForwardQueueCount: 0,
+ currentTotalLag: 30,
+ projectedTotalLag: 40,
+ totalOffsetDelta: -10,
+ warnings: ['1 queue(s) will move backward and may replay consumed
messages'],
+ queues: [
+ {
+ topic: 'remote-topic',
+ broker: 'broker-a',
+ queueId: 0,
+ minOffset: 0,
+ maxOffset: 200,
+ brokerOffset: 120,
+ consumerOffset: 90,
+ targetOffset: 80,
+ currentLag: 30,
+ projectedLag: 40,
+ offsetDelta: -10,
+ riskLevel: 'WARNING',
+ message: 'Replays 10 message(s)',
+ },
+ ],
+ });
vi.mocked(consumerService.getConsumerStack).mockResolvedValue({
groupName: 'remote-cg',
clientId: 'client-1',
@@ -250,6 +288,12 @@ describe('Consumer page', () => {
});
});
+ afterEach(() => {
+ cleanup();
+ Modal.destroyAll();
+ message.destroy();
+ });
+
it('submits the canonical global delivery order type', async () => {
const user = userEvent.setup();
const confirmSpy = vi.spyOn(Modal, 'confirm').mockImplementation((config)
=> {
@@ -469,7 +513,7 @@ describe('Consumer page', () => {
);
});
- it('requires and submits a target topic when resetting consumer offsets',
async () => {
+ it('previews queue impact before resetting consumer offsets', async () => {
const user = userEvent.setup();
renderWithProviders(<ConsumerPage />);
@@ -494,6 +538,22 @@ describe('Consumer page', () => {
return element;
});
await user.click(option);
+ expect(confirm).toBeDisabled();
+
+ await user.click(screen.getByRole('button', { name: /预览影响/ }));
+
+ await waitFor(() =>
+ expect(consumerService.previewConsumerOffsetReset).toHaveBeenCalledWith(
+ expect.objectContaining({
+ name: 'remote-cg',
+ instanceId: 'instance-1',
+ topic: 'remote-topic',
+ timestamp: expect.any(Number),
+ }),
+ ),
+ );
+ expect(await screen.findByText('重置后总堆积')).toBeInTheDocument();
+ expect(await screen.findByText('将回放 10 条消息')).toBeInTheDocument();
await waitFor(() => expect(confirm).toBeEnabled());
await user.click(confirm);
@@ -509,6 +569,87 @@ describe('Consumer page', () => {
);
});
+ it('invalidates the reset preview when reset parameters change', async () =>
{
+ const user = userEvent.setup();
+ renderWithProviders(<ConsumerPage />);
+
+ await user.click(await screen.findByRole('button', { name: /重置位点/ }));
+ const topicSelect = screen.getByRole('combobox', { name: '目标 Topic' });
+ await user.click(topicSelect);
+ 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: /预览影响/ }));
+
+ expect(await screen.findByText('将回放 10 条消息')).toBeInTheDocument();
+ await waitFor(() => expect(screen.getByRole('button', { name: '确认重置'
})).toBeEnabled());
+
+ await user.click(screen.getByRole('button', { name: '1 小时前' }));
+
+ expect(screen.queryByText('将回放 10 条消息')).not.toBeInTheDocument();
+ expect(screen.getByRole('button', { name: '确认重置' })).toBeDisabled();
+ });
+
+ it('blocks reset confirmation when the preview has failed queues', async ()
=> {
+ vi.mocked(consumerService.previewConsumerOffsetReset).mockResolvedValue({
+ instanceId: 'instance-1',
+ groupName: 'remote-cg',
+ topic: 'remote-topic',
+ timestamp: 1784246400000,
+ complete: false,
+ allowReset: false,
+ queueCount: 1,
+ warningCount: 1,
+ rewindQueueCount: 0,
+ fastForwardQueueCount: 0,
+ currentTotalLag: 30,
+ projectedTotalLag: 30,
+ totalOffsetDelta: 0,
+ warnings: ['Failed to preview 1 queue(s); retry before applying the
reset'],
+ queues: [
+ {
+ topic: 'remote-topic',
+ broker: 'broker-a',
+ queueId: 0,
+ minOffset: -1,
+ maxOffset: -1,
+ brokerOffset: 120,
+ consumerOffset: 90,
+ targetOffset: 90,
+ currentLag: 30,
+ projectedLag: 30,
+ offsetDelta: 0,
+ riskLevel: 'ERROR',
+ message: 'Failed to preview queue offset: timeout',
+ },
+ ],
+ });
+ const user = userEvent.setup();
+ 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: /预览影响/ }));
+
+ expect(await screen.findByText('不完整')).toBeInTheDocument();
+ expect(await screen.findByText('Failed to preview queue offset:
timeout')).toBeInTheDocument();
+ expect(screen.getByRole('button', { name: '确认重置' })).toBeDisabled();
+ 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 a5c1cf993..bed410356 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -72,6 +72,8 @@ import type {
ConsumerInstance,
ConsumerStackTrace,
QueueProgress,
+ ResetConsumerOffsetPreview,
+ ResetConsumerOffsetQueuePreview,
SubscriptionEntry,
} from '../../api/metadata';
import {
@@ -84,6 +86,7 @@ import {
getConsumerSubscriptions,
importConsumerGroups,
listConsumerGroupPage,
+ previewConsumerOffsetReset,
refreshConsumerGroup,
resetConsumerOffset,
getConsumerGroupSettings,
@@ -165,6 +168,50 @@ const isConsistentSubscription = (subscription:
SubscriptionEntry): boolean =>
const isInconsistentSubscription = (subscription: SubscriptionEntry): boolean
=>
isInconsistentValue(subscription.consistency);
+const formatOffsetValue = (value: number) =>
+ Number.isFinite(value) && value >= 0 ? value.toLocaleString() : '-';
+
+const formatOffsetDelta = (value: number) => {
+ if (value > 0) return `+${value.toLocaleString()}`;
+ return value.toLocaleString();
+};
+
+const resetPreviewRiskColor = (riskLevel: string) => {
+ if (riskLevel === 'ERROR') return 'red';
+ if (riskLevel === 'WARNING') return 'orange';
+ return 'green';
+};
+
+const resetPreviewRiskLabel = (riskLevel: string) => {
+ if (riskLevel === 'ERROR') return '失败';
+ if (riskLevel === 'WARNING') return '需确认';
+ return '正常';
+};
+
+const resetPreviewQueueMessage = (queue: ResetConsumerOffsetQueuePreview) => {
+ const messages: string[] = [];
+ if (queue.riskLevel === 'ERROR') {
+ return queue.message || '预览失败';
+ }
+ if (queue.targetOffset < 0 || queue.consumerOffset < 0) {
+ return queue.message || '目标位点不可用';
+ }
+ if (queue.offsetDelta < 0) {
+ messages.push(`将回放 ${Math.abs(queue.offsetDelta).toLocaleString()} 条消息`);
+ } else if (queue.offsetDelta > 0) {
+ messages.push(`将跳过 ${queue.offsetDelta.toLocaleString()} 条未消费消息`);
+ } else {
+ messages.push('位点不变');
+ }
+ if (queue.minOffset >= 0 && queue.targetOffset === queue.minOffset) {
+ messages.push('目标为最小保留位点');
+ }
+ if (queue.maxOffset >= 0 && queue.targetOffset === queue.maxOffset) {
+ messages.push('目标为最新位点');
+ }
+ return messages.join('；');
+};
+
// 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) =>
@@ -210,6 +257,10 @@ const ConsumerPageContent = ({
const [resetGroup, setResetGroup] = useState<ConsumerGroup | null>(null);
const [resetTopic, setResetTopic] = useState<string>();
const [resetTime, setResetTime] = useState<Dayjs>(dayjs().subtract(3,
'hour'));
+ const [resetPreview, setResetPreview] = useState<ResetConsumerOffsetPreview
| null>(null);
+ const [resetPreviewKey, setResetPreviewKey] = useState('');
+ const [resetPreviewLoading, setResetPreviewLoading] = useState(false);
+ const [resetPreviewError, setResetPreviewError] = useState<string |
null>(null);
const [subscriptionsByGroup, setSubscriptionsByGroup] = useState<
Record<string, SubscriptionEntry[]>
>({});
@@ -245,6 +296,11 @@ const ConsumerPageContent = ({
silentRefreshRef.current = silent;
setRefreshKey((key) => key + 1);
}, []);
+ const clearResetPreview = useCallback(() => {
+ setResetPreview(null);
+ setResetPreviewKey('');
+ setResetPreviewError(null);
+ }, []);
const selectedGroupName = selectedGroup?.name;
useEffect(() => {
@@ -334,8 +390,8 @@ const ConsumerPageContent = ({
);
useEffect(() => {
- if (!modalOpen || !selectedGroup || !selectedInstanceId) return undefined;
- const groupName = selectedGroup.name;
+ if (!modalOpen || !selectedGroupName || !selectedInstanceId) return
undefined;
+ const groupName = selectedGroupName;
let inFlight = false;
const tick = async () => {
if (inFlight) return;
@@ -355,7 +411,7 @@ const ConsumerPageContent = ({
};
const interval = window.setInterval(() => void tick(), 2000);
return () => window.clearInterval(interval);
- }, [modalOpen, selectedGroup?.name, selectedInstanceId, loadProgress]);
+ }, [modalOpen, selectedGroupName, selectedInstanceId, loadProgress]);
/* ─── Filtered & sorted data ─── */
const filtered = useMemo(() => {
@@ -456,6 +512,17 @@ const ConsumerPageContent = ({
const resetDiagnosticKey = resetGroup
? diagnosticCacheKey(selectedInstanceId, resetGroup.name)
: '';
+ const resetTimestamp = resetTime.valueOf();
+ const currentResetPreviewKey =
+ resetGroup && resetTopic
+ ? [selectedInstanceId ?? '', resetGroup.name, resetTopic,
resetTimestamp].join('\u0000')
+ : '';
+ const hasCurrentResetPreview = Boolean(
+ resetPreview && resetPreviewKey === currentResetPreviewKey,
+ );
+ const resetPreviewQueues = hasCurrentResetPreview ? (resetPreview?.queues ??
[]) : [];
+ const resetPreviewWarnings = hasCurrentResetPreview ?
(resetPreview?.warnings ?? []) : [];
+ const resetPreviewCanApply = Boolean(hasCurrentResetPreview &&
resetPreview?.allowReset);
const resetTopicOptions = useMemo(() => {
const topics = new Set(resetGroup?.subscribedTopics ?? []);
for (const subscription of subscriptionsByGroup[resetDiagnosticKey] ?? [])
{
@@ -501,6 +568,73 @@ const ConsumerPageContent = ({
0,
);
+ const handlePreviewResetOffset = async () => {
+ if (!resetGroup || !resetTopic) {
+ message.warning('请先选择要重置的 Topic');
+ return;
+ }
+ const previewKey = currentResetPreviewKey;
+ setResetPreviewLoading(true);
+ setResetPreviewError(null);
+ try {
+ const preview = await previewConsumerOffsetReset({
+ name: resetGroup.name,
+ instanceId: selectedInstanceId || undefined,
+ topic: resetTopic,
+ timestamp: resetTimestamp,
+ });
+ setResetPreview(preview);
+ setResetPreviewKey(previewKey);
+ if (preview.complete && preview.queueCount > 0) {
+ message.success(`已预览 ${preview.queueCount} 个 Queue`);
+ } else {
+ message.warning('预览未覆盖可重置队列，请检查 Group/Topic 状态');
+ }
+ } catch (error) {
+ const reason = error instanceof Error ? error.message : '预览重置影响失败';
+ setResetPreview(null);
+ setResetPreviewKey('');
+ setResetPreviewError(reason);
+ message.error(reason);
+ } finally {
+ setResetPreviewLoading(false);
+ }
+ };
+
+ const handleResetOffset = async () => {
+ if (!resetGroup || !resetTopic) return;
+ if (!resetPreviewCanApply) {
+ message.warning('请先预览并确认位点影响');
+ return;
+ }
+ setResetSubmitting(true);
+ try {
+ await resetConsumerOffset({
+ name: resetGroup.name,
+ instanceId: selectedInstanceId || undefined,
+ topic: resetTopic,
+ timestamp: resetTimestamp,
+ });
+ message.success(
+ `${resetGroup.name} 在 ${resetTopic} 的消费位点已重置到
${resetTime.format('YYYY-MM-DD HH:mm:ss')}`,
+ );
+ setProgressByGroup((prev) => {
+ const next = { ...prev };
+ delete next[resetDiagnosticKey];
+ return next;
+ });
+ triggerRefresh(true);
+ setResetModalOpen(false);
+ setResetGroup(null);
+ setResetTopic(undefined);
+ clearResetPreview();
+ } catch {
+ message.error(t('consumer.resetFailed'));
+ } finally {
+ setResetSubmitting(false);
+ }
+ };
+
const openStackModal = async (consumerInstance: ConsumerInstance) => {
if (!selectedGroup) return;
const requestId = ++stackRequestIdRef.current;
@@ -784,6 +918,7 @@ const ConsumerPageContent = ({
setResetGroup(record);
setResetTopic(undefined);
setResetTime(dayjs().subtract(3, 'hour'));
+ clearResetPreview();
setResetModalOpen(true);
void loadSubscriptions(record.name);
}}
@@ -1036,6 +1171,111 @@ const ConsumerPageContent = ({
},
];
+ const resetPreviewColumns: ColumnsType<ResetConsumerOffsetQueuePreview> = [
+ {
+ title: 'Broker',
+ dataIndex: 'broker',
+ key: 'broker',
+ width: 140,
+ ellipsis: true,
+ render: (broker: string) => (
+ <Text strong style={{ fontSize: 14 }} title={broker}>
+ {broker || '-'}
+ </Text>
+ ),
+ },
+ {
+ title: 'Queue ID',
+ dataIndex: 'queueId',
+ key: 'queueId',
+ width: 86,
+ align: 'center',
+ render: (id: number) => <Tag color="blue">Queue {id}</Tag>,
+ },
+ {
+ title: '当前位点',
+ dataIndex: 'consumerOffset',
+ key: 'consumerOffset',
+ width: 120,
+ align: 'right',
+ render: (offset: number) => (
+ <Text style={{ fontFamily: 'monospace'
}}>{formatOffsetValue(offset)}</Text>
+ ),
+ },
+ {
+ title: '目标位点',
+ dataIndex: 'targetOffset',
+ key: 'targetOffset',
+ width: 120,
+ align: 'right',
+ render: (offset: number) => (
+ <Text strong style={{ fontFamily: 'monospace' }}>
+ {formatOffsetValue(offset)}
+ </Text>
+ ),
+ },
+ {
+ title: '变化',
+ dataIndex: 'offsetDelta',
+ key: 'offsetDelta',
+ width: 100,
+ align: 'right',
+ render: (delta: number, row: ResetConsumerOffsetQueuePreview) => (
+ <Text
+ style={{
+ color: delta > 0 ? '#fa8c16' : delta < 0 ? '#1677ff' : undefined,
+ fontFamily: 'monospace',
+ fontWeight: 600,
+ }}
+ >
+ {row.targetOffset < 0 || row.consumerOffset < 0 ? '-' :
formatOffsetDelta(delta)}
+ </Text>
+ ),
+ },
+ {
+ title: '当前堆积',
+ dataIndex: 'currentLag',
+ key: 'currentLag',
+ width: 110,
+ align: 'right',
+ render: (lag: number) => (
+ <Text style={{ fontFamily: 'monospace'
}}>{formatOffsetValue(lag)}</Text>
+ ),
+ },
+ {
+ title: '重置后堆积',
+ dataIndex: 'projectedLag',
+ key: 'projectedLag',
+ width: 124,
+ align: 'right',
+ render: (lag: number) => (
+ <Text style={{ fontFamily: 'monospace', color: lagColor(lag),
fontWeight: 600 }}>
+ {formatOffsetValue(lag)}
+ </Text>
+ ),
+ },
+ {
+ title: '风险',
+ dataIndex: 'riskLevel',
+ key: 'riskLevel',
+ width: 86,
+ render: (riskLevel: string) => (
+ <Tag
color={resetPreviewRiskColor(riskLevel)}>{resetPreviewRiskLabel(riskLevel)}</Tag>
+ ),
+ },
+ {
+ title: '说明',
+ key: 'message',
+ width: 240,
+ ellipsis: true,
+ render: (_: unknown, record: ResetConsumerOffsetQueuePreview) => (
+ <Text style={{ fontSize: 14 }}
title={resetPreviewQueueMessage(record)}>
+ {resetPreviewQueueMessage(record)}
+ </Text>
+ ),
+ },
+ ];
+
/* ═══════════════════════════════════════════
Render
═══════════════════════════════════════════ */
@@ -1919,54 +2159,29 @@ const ConsumerPageContent = ({
setResetModalOpen(false);
setResetGroup(null);
setResetTopic(undefined);
+ clearResetPreview();
}}
- onOk={async () => {
- if (!resetGroup || !resetTopic) return;
- setResetSubmitting(true);
- try {
- await resetConsumerOffset({
- name: resetGroup.name,
- instanceId: selectedInstanceId || undefined,
- topic: resetTopic,
- timestamp: resetTime.valueOf(),
- });
- message.success(
- `${resetGroup.name} 在 ${resetTopic} 的消费位点已重置到
${resetTime.format('YYYY-MM-DD HH:mm:ss')}`,
- );
- } catch {
- message.error(t('consumer.resetFailed'));
- return;
- } finally {
- setResetSubmitting(false);
- }
- setResetModalOpen(false);
- setResetGroup(null);
- setResetTopic(undefined);
- }}
+ onOk={() => void handleResetOffset()}
confirmLoading={resetSubmitting}
okButtonProps={{
- disabled: !resetTopic ||
Boolean(subscriptionLoadingByGroup[resetDiagnosticKey]),
+ disabled:
+ !resetPreviewCanApply ||
+ resetPreviewLoading ||
+ Boolean(subscriptionLoadingByGroup[resetDiagnosticKey]),
}}
okText="确认重置"
cancelText="取消"
- width={480}
+ width={1200}
destroyOnHidden
>
{resetGroup && (
- <div style={{ marginTop: 16 }}>
- <div
- style={{
- marginBottom: 16,
- padding: '12px 16px',
- background: '#fff7e6',
- borderRadius: 8,
- border: '1px solid #ffd591',
- }}
- >
- <Text type="warning" style={{ fontSize: 14 }}>
- ⚠️ 此操作将影响消息消费进度，请谨慎操作。重置后消费者将从指定时间点开始重新消费。
- </Text>
- </div>
+ <Space direction="vertical" size={16} style={{ width: '100%',
marginTop: 16 }}>
+ <Alert
+ showIcon
+ type="warning"
+ message="此操作将影响消息消费进度"
+ description="请先预览每个 Queue
的目标位点和堆积变化，确认预览结果后再执行重置。预览为时点快照、属页面操作引导（非服务端控制），预览期间消息持续写入，实际效果以执行时 Broker
状态为准。"
+ />
<div style={{ marginBottom: 16 }}>
<Text type="secondary" style={{ fontSize: 14, display: 'block',
marginBottom: 4 }}>
目标 Group
@@ -1988,7 +2203,10 @@ const ConsumerPageContent = ({
options={resetTopicOptions}
loading={subscriptionLoadingByGroup[resetDiagnosticKey]}
placeholder="选择要重置消费位点的 Topic"
- onChange={setResetTopic}
+ onChange={(value) => {
+ setResetTopic(value);
+ clearResetPreview();
+ }}
notFoundContent={
subscriptionErrorByGroup[resetDiagnosticKey]
? '订阅 Topic 加载失败'
@@ -2005,7 +2223,10 @@ const ConsumerPageContent = ({
style={{ width: '100%' }}
value={resetTime}
onChange={(val) => {
- if (val) setResetTime(val);
+ if (val) {
+ setResetTime(val);
+ clearResetPreview();
+ }
}}
format="YYYY-MM-DD HH:mm:ss"
placeholder="选择重置时间点"
@@ -2016,27 +2237,131 @@ const ConsumerPageContent = ({
快捷选择
</Text>
<Space wrap>
- <Button size="small" onClick={() =>
setResetTime(dayjs().subtract(1, 'hour'))}>
+ <Button
+ size="small"
+ onClick={() => {
+ setResetTime(dayjs().subtract(1, 'hour'));
+ clearResetPreview();
+ }}
+ >
1 小时前
</Button>
- <Button size="small" onClick={() =>
setResetTime(dayjs().subtract(3, 'hour'))}>
+ <Button
+ size="small"
+ onClick={() => {
+ setResetTime(dayjs().subtract(3, 'hour'));
+ clearResetPreview();
+ }}
+ >
3 小时前
</Button>
- <Button size="small" onClick={() =>
setResetTime(dayjs().subtract(6, 'hour'))}>
+ <Button
+ size="small"
+ onClick={() => {
+ setResetTime(dayjs().subtract(6, 'hour'));
+ clearResetPreview();
+ }}
+ >
6 小时前
</Button>
- <Button size="small" onClick={() =>
setResetTime(dayjs().subtract(12, 'hour'))}>
+ <Button
+ size="small"
+ onClick={() => {
+ setResetTime(dayjs().subtract(12, 'hour'));
+ clearResetPreview();
+ }}
+ >
12 小时前
</Button>
- <Button size="small" onClick={() =>
setResetTime(dayjs().subtract(1, 'day'))}>
+ <Button
+ size="small"
+ onClick={() => {
+ setResetTime(dayjs().subtract(1, 'day'));
+ clearResetPreview();
+ }}
+ >
1 天前
</Button>
- <Button size="small" onClick={() =>
setResetTime(dayjs().subtract(3, 'day'))}>
+ <Button
+ size="small"
+ onClick={() => {
+ setResetTime(dayjs().subtract(3, 'day'));
+ clearResetPreview();
+ }}
+ >
3 天前
</Button>
</Space>
</div>
- </div>
+ <Flex justify="space-between" align="center" gap={12}>
+ <Text type="secondary" style={{ fontSize: 14 }}>
+ 预览不会修改 broker 位点，仅计算目标时间对应的 Queue offset。
+ </Text>
+ <Button
+ icon={<Eye size={14} />}
+ loading={resetPreviewLoading}
+ disabled={
+ !resetTopic ||
+ Boolean(subscriptionLoadingByGroup[resetDiagnosticKey]) ||
+ resetSubmitting
+ }
+ onClick={() => void handlePreviewResetOffset()}
+ >
+ 预览影响
+ </Button>
+ </Flex>
+ {resetPreviewError && (
+ <Alert showIcon type="error" message="位点预览失败"
description={resetPreviewError} />
+ )}
+ {hasCurrentResetPreview && resetPreview && (
+ <Space direction="vertical" size={12} style={{ width: '100%' }}>
+ <Descriptions bordered size="small" column={4}>
+ <Descriptions.Item label="Queue
数">{resetPreview.queueCount}</Descriptions.Item>
+ <Descriptions.Item label="当前总堆积">
+ {formatOffsetValue(resetPreview.currentTotalLag)}
+ </Descriptions.Item>
+ <Descriptions.Item label="重置后总堆积">
+ {formatOffsetValue(resetPreview.projectedTotalLag)}
+ </Descriptions.Item>
+ <Descriptions.Item label="位点净变化">
+ {formatOffsetDelta(resetPreview.totalOffsetDelta)}
+ </Descriptions.Item>
+ <Descriptions.Item label="回放 Queue">
+ {resetPreview.rewindQueueCount}
+ </Descriptions.Item>
+ <Descriptions.Item label="跳过 Queue">
+ {resetPreview.fastForwardQueueCount}
+ </Descriptions.Item>
+ <Descriptions.Item label="预览状态" span={2}>
+ <Tag
+ color={
+ resetPreview.complete ? 'green' :
resetPreview.allowReset ? 'orange' : 'red'
+ }
+ >
+ {resetPreview.complete ? '完整' : resetPreview.allowReset
? '有限' : '不完整'}
+ </Tag>
+ </Descriptions.Item>
+ </Descriptions>
+ {resetPreviewWarnings.length > 0 && (
+ <Alert
+ showIcon
+ type={resetPreview.complete ? 'warning' : 'error'}
+ message="请确认以下影响"
+ description={resetPreviewWarnings.join('；')}
+ />
+ )}
+ <Table
+ columns={resetPreviewColumns}
+ dataSource={resetPreviewQueues}
+ rowKey={(row) => `${row.topic}-${row.broker}-${row.queueId}`}
+ pagination={false}
+ size="small"
+ scroll={{ x: tableScrollX(resetPreviewColumns), y: 260 }}
+ locale={{ emptyText: '未找到可预览的 Queue 位点' }}
+ />
+ </Space>
+ )}
+ </Space>
)}
</Modal>
</div>
diff --git a/web/src/services/consumerService.test.ts
b/web/src/services/consumerService.test.ts
index 897e19e3f..44463d470 100644
--- a/web/src/services/consumerService.test.ts
+++ b/web/src/services/consumerService.test.ts
@@ -25,6 +25,7 @@ import {
listAllConsumerGroups,
listConsumerGroupPage,
listConsumerGroups,
+ previewConsumerOffsetReset,
} from './consumerService';
const { mode, metadataApi } = vi.hoisted(() => ({
@@ -33,6 +34,7 @@ const { mode, metadataApi } = vi.hoisted(() => ({
getConsumerGroup: vi.fn(),
listConsumerGroupPage: vi.fn(),
listConsumerGroups: vi.fn(),
+ previewConsumerOffsetReset: vi.fn(),
},
}));
@@ -187,6 +189,67 @@ describe('consumer service mock data', () => {
expect(secondSubscriptions[0]).not.toBe(firstSubscriptions[0]);
});
+ it('builds copied reset offset previews in mock mode', async () => {
+ const firstPreview = await previewConsumerOffsetReset({
+ name: 'cg-order-notify',
+ instanceId: 'instance-proxy-1',
+ topic: 'order-create',
+ timestamp: 1784246400000,
+ });
+
+ expect(firstPreview.groupName).toBe('cg-order-notify');
+ expect(firstPreview.topic).toBe('order-create');
+ expect(firstPreview.complete).toBe(true);
+ expect(firstPreview.queueCount).toBeGreaterThan(0);
+ expect(firstPreview.rewindQueueCount).toBeGreaterThan(0);
+ expect(firstPreview.queues[0].topic).toBe('order-create');
+ firstPreview.queues[0].broker = 'mutated-broker';
+ firstPreview.warnings.push('mutated-warning');
+
+ const secondPreview = await previewConsumerOffsetReset({
+ name: 'cg-order-notify',
+ instanceId: 'instance-proxy-1',
+ topic: 'order-create',
+ timestamp: 1784246400000,
+ });
+ expect(secondPreview.queues[0].broker).not.toBe('mutated-broker');
+ expect(secondPreview.warnings).not.toContain('mutated-warning');
+ });
+
+ it('forwards reset offset previews in API mode', async () => {
+ mode.mock = false;
+ const request = {
+ name: 'cg-orders',
+ instanceId: 'instance-1',
+ topic: 'orders',
+ timestamp: 1784246400000,
+ };
+ const preview = {
+ instanceId: 'instance-1',
+ groupName: 'cg-orders',
+ topic: 'orders',
+ timestamp: 1784246400000,
+ complete: true,
+ allowReset: true,
+ queueCount: 0,
+ warningCount: 0,
+ rewindQueueCount: 0,
+ fastForwardQueueCount: 0,
+ currentTotalLag: 0,
+ projectedTotalLag: 0,
+ totalOffsetDelta: 0,
+ warnings: [],
+ queues: [],
+ };
+ metadataApi.previewConsumerOffsetReset.mockResolvedValue(preview);
+ try {
+ await
expect(previewConsumerOffsetReset(request)).resolves.toEqual(preview);
+
expect(metadataApi.previewConsumerOffsetReset).toHaveBeenCalledWith(request);
+ } finally {
+ mode.mock = true;
+ }
+ });
+
it('returns an empty mock consumer stack trace', async () => {
const stack = await getConsumerStack('cg-order-notify', 'client-1');
diff --git a/web/src/services/consumerService.ts
b/web/src/services/consumerService.ts
index 24d9767a2..098b0de67 100644
--- a/web/src/services/consumerService.ts
+++ b/web/src/services/consumerService.ts
@@ -11,6 +11,8 @@ import type {
ImportConsumerGroupsResult,
PageResult,
QueueProgress,
+ ResetConsumerOffsetPreview,
+ ResetConsumerOffsetQueuePreview,
ResetConsumerOffsetRequest,
SubscriptionEntry,
} from '../api/metadata';
@@ -63,6 +65,14 @@ function copySubscription(subscription: SubscriptionEntry):
SubscriptionEntry {
return { ...subscription };
}
+function copyResetOffsetPreview(preview: ResetConsumerOffsetPreview):
ResetConsumerOffsetPreview {
+ return {
+ ...preview,
+ warnings: [...preview.warnings],
+ queues: preview.queues.map((queue) => ({ ...queue })),
+ };
+}
+
const normalizeConsumerGroup = <T extends ConsumerGroup>(group: T): T => ({
...group,
subscribedTopics: group.subscribedTopics ?? [],
@@ -296,6 +306,96 @@ export async function resetConsumerOffset(data:
ResetConsumerOffsetRequest): Pro
return metadataApi.resetConsumerOffset(data);
}
+export async function previewConsumerOffsetReset(
+ data: ResetConsumerOffsetRequest,
+): Promise<ResetConsumerOffsetPreview> {
+ if (isMockMode()) return buildMockResetOffsetPreview(data);
+ return metadataApi.previewConsumerOffsetReset(data);
+}
+
+function buildMockResetOffsetPreview(data: ResetConsumerOffsetRequest):
ResetConsumerOffsetPreview {
+ const progressRows = ((mockQueueProgress[data.name] as unknown as
QueueProgress[]) ?? []).filter(
+ (progress) => !progress.topic || progress.topic === data.topic,
+ );
+ const queues = progressRows.map((progress) =>
+ buildMockResetOffsetQueuePreview(data.topic, progress),
+ );
+ const currentTotalLag = queues.reduce((sum, queue) => sum +
queue.currentLag, 0);
+ const projectedTotalLag = queues.reduce((sum, queue) => sum +
queue.projectedLag, 0);
+ const rewindQueueCount = queues.filter((queue) => queue.offsetDelta <
0).length;
+ const fastForwardQueueCount = queues.filter((queue) => queue.offsetDelta >
0).length;
+ const warnings = buildMockResetOffsetWarnings(queues, rewindQueueCount,
fastForwardQueueCount);
+
+ return copyResetOffsetPreview({
+ instanceId: data.instanceId,
+ groupName: data.name,
+ topic: data.topic,
+ timestamp: data.timestamp,
+ complete: queues.length > 0,
+ allowReset: queues.length > 0,
+ queueCount: queues.length,
+ warningCount: warnings.length,
+ rewindQueueCount,
+ fastForwardQueueCount,
+ currentTotalLag,
+ projectedTotalLag,
+ totalOffsetDelta: queues.reduce((sum, queue) => sum + queue.offsetDelta,
0),
+ warnings,
+ queues,
+ });
+}
+
+function buildMockResetOffsetQueuePreview(
+ topic: string,
+ progress: QueueProgress,
+): ResetConsumerOffsetQueuePreview {
+ const maxOffset = progress.brokerOffset;
+ const minOffset = 0;
+ const rewindWindow = Math.min(500, Math.max(1,
Math.floor(Math.max(progress.diffTotal, 1) / 2)));
+ const targetOffset = Math.max(
+ minOffset,
+ Math.min(maxOffset, progress.consumerOffset - rewindWindow),
+ );
+ const offsetDelta = targetOffset - progress.consumerOffset;
+ const currentLag = Math.max(0, progress.brokerOffset -
progress.consumerOffset);
+ const projectedLag = Math.max(0, progress.brokerOffset - targetOffset);
+
+ return {
+ topic: progress.topic || topic,
+ broker: progress.broker,
+ queueId: progress.queueId,
+ minOffset,
+ maxOffset,
+ brokerOffset: progress.brokerOffset,
+ consumerOffset: progress.consumerOffset,
+ targetOffset,
+ currentLag,
+ projectedLag,
+ offsetDelta,
+ riskLevel: offsetDelta === 0 ? 'INFO' : 'WARNING',
+ message:
+ offsetDelta === 0
+ ? 'Offset unchanged'
+ : `Replays ${Math.abs(offsetDelta).toLocaleString()} message(s)`,
+ };
+}
+
+function buildMockResetOffsetWarnings(
+ queues: ResetConsumerOffsetQueuePreview[],
+ rewindQueueCount: number,
+ fastForwardQueueCount: number,
+): string[] {
+ if (queues.length === 0) return ['No consume offset data found for the
selected topic'];
+ const warnings: string[] = [];
+ if (fastForwardQueueCount > 0) {
+ warnings.push(`${fastForwardQueueCount} queue(s) will move forward and may
skip messages`);
+ }
+ if (rewindQueueCount > 0) {
+ warnings.push(`${rewindQueueCount} queue(s) will replay consumed
messages`);
+ }
+ return warnings;
+}
+
export interface BatchDeleteConsumerGroupsResult {
deleted: string[];
failed: string[];
