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 9e0ed86e5 feat(studio): update consumer group settings (#2528)
9e0ed86e5 is described below
commit 9e0ed86e5d567aaa2ba89e869ff3170387703bc5
Author: aias00 <[email protected]>
AuthorDate: Sat Aug 22 15:37:05 2026 +0800
feat(studio): update consumer group settings (#2528)
---
.../instance/group/ConsumerGroupController.java | 13 ++++
.../instance/group/ConsumerGroupSettingsVO.java | 32 ++++++++
.../group/UpdateConsumerGroupSettingsDTO.java | 36 +++++++++
.../studio/instance/topic/MetadataService.java | 21 +++++
.../studio/provider/apache/AdminClient.java | 4 +
.../provider/apache/RocketMQAdminClientImpl.java | 65 ++++++++++++++++
.../group/ConsumerGroupControllerTest.java | 29 +++++++
.../apache/RocketMQAdminClientImplTest.java | 39 ++++++++++
web/src/api/consumerGroups.test.ts | 24 ++++++
web/src/api/metadata.ts | 21 +++++
web/src/pages/instance/consumer.tsx | 90 ++++++++++++++++++++++
web/src/services/consumerService.ts | 21 +++++
12 files changed, 395 insertions(+)
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 bf7988d31..d6d0733ca 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
@@ -66,6 +66,19 @@ public class ConsumerGroupController {
return Result.ok(metadataService.getConsumerGroup(instanceId, name));
}
+ @GetMapping("/{name}/settings")
+ public Result<ConsumerGroupSettingsVO>
getConsumerGroupSettings(@PathVariable String name,
+
@RequestParam String instanceId) {
+ return Result.ok(metadataService.getConsumerGroupSettings(instanceId,
name));
+ }
+
+ @PostMapping("/settings")
+ public Result<ConsumerGroupSettingsVO> updateConsumerGroupSettings(
+ @Valid @RequestBody UpdateConsumerGroupSettingsDTO request) {
+ return
Result.ok(metadataService.updateConsumerGroupSettings(request.getInstanceId(),
request.getName(),
+ request.getRetryQueueNums(), request.getRetryMaxTimes()));
+ }
+
@GetMapping("/{name}/progress")
public Result<List<QueueProgressVO>> getGroupProgress(
@PathVariable String name,
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupSettingsVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupSettingsVO.java
new file mode 100644
index 000000000..7cbff0e5e
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupSettingsVO.java
@@ -0,0 +1,32 @@
+/*
+ * 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 ConsumerGroupSettingsVO {
+ private String groupName;
+ private int retryQueueNums;
+ private int retryMaxTimes;
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/UpdateConsumerGroupSettingsDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/UpdateConsumerGroupSettingsDTO.java
new file mode 100644
index 000000000..0c3cf330f
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/UpdateConsumerGroupSettingsDTO.java
@@ -0,0 +1,36 @@
+/*
+ * 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 jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Positive;
+import lombok.Data;
+
+@Data
+public class UpdateConsumerGroupSettingsDTO {
+ @NotBlank(message = "instanceId is required")
+ private String instanceId;
+ @NotBlank(message = "name is required")
+ private String name;
+ @NotNull(message = "retryQueueNums is required")
+ @Positive(message = "retryQueueNums must be positive")
+ private Integer retryQueueNums;
+ @NotNull(message = "retryMaxTimes is required")
+ @Positive(message = "retryMaxTimes must be positive")
+ private Integer retryMaxTimes;
+}
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 bfe9a6a44..17d492a34 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
@@ -24,6 +24,7 @@ import
org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
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.SubscriptionEntryVO;
import org.apache.rocketmq.studio.provider.InstanceProvider;
@@ -235,6 +236,20 @@ public class MetadataService {
return resolve(instanceId).createConsumerGroup(instanceId, group);
}
+ public ConsumerGroupSettingsVO getConsumerGroupSettings(String instanceId,
String name) {
+ instanceId = normalizeInstanceId(instanceId);
+ requireApacheInstance(instanceId);
+ return adminClient.getConsumerGroupSettings(instanceId,
requireName(name, "consumer group name"));
+ }
+
+ public ConsumerGroupSettingsVO updateConsumerGroupSettings(String
instanceId, String name, int retryQueueNums,
+ int
retryMaxTimes) {
+ instanceId = normalizeInstanceId(instanceId);
+ requireApacheInstance(instanceId);
+ return adminClient.updateConsumerGroupSettings(instanceId,
requireName(name, "consumer group name"),
+ retryQueueNums, retryMaxTimes);
+ }
+
public void deleteConsumerGroup(String name) {
deleteConsumerGroup(null, name);
@@ -265,6 +280,12 @@ public class MetadataService {
.orElseGet(() ->
providerRegistry.forVendor(InstanceVendor.APACHE));
}
+ private void requireApacheInstance(String instanceId) {
+ if (resolve(instanceId).vendor() != InstanceVendor.APACHE) {
+ throw new BusinessException(501, "Consumer group settings are not
supported for cloud instances");
+ }
+ }
+
private String normalizeFilter(String value) {
return !StringUtils.hasText(value) ? null : value.trim();
}
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 8d65f1266..36ecea467 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
@@ -20,6 +20,7 @@ import
org.apache.rocketmq.studio.instance.topic.SendMessageVO;
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;
public interface AdminClient {
@@ -30,6 +31,9 @@ public interface AdminClient {
void deleteTopic(String instanceId, String name);
SendMessageVO sendMessage(SendMessageDTO request);
ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group);
+ ConsumerGroupSettingsVO getConsumerGroupSettings(String instanceId, String
name);
+ ConsumerGroupSettingsVO updateConsumerGroupSettings(String instanceId,
String name, int retryQueueNums,
+ int retryMaxTimes);
void deleteConsumerGroup(String instanceId, String name);
void resetOffset(String instanceId, String name, long timestamp, String
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 3190569f9..8c223a3d8 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
@@ -37,6 +37,7 @@ import
org.apache.rocketmq.studio.common.exception.BusinessException;
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.topic.SendMessageDTO;
import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
import org.apache.rocketmq.studio.instance.topic.TopicVO;
@@ -499,6 +500,70 @@ public class RocketMQAdminClientImpl implements
AdminClient {
return adminFactory.execute(namesrvAddr(), null, admin ->
createConsumerGroup(admin, group));
}
+ @Override
+ public ConsumerGroupSettingsVO getConsumerGroupSettings(String instanceId,
String name) {
+ return executeForInstance(instanceId, admin -> {
+ try {
+ Set<String> brokerAddrs =
getMasterBrokerAddrsForCluster(admin, getClusterName(admin));
+ if (brokerAddrs.isEmpty()) {
+ throw new BusinessException(502, "No broker available to
get consumer group settings");
+ }
+ SubscriptionGroupConfig config =
admin.examineSubscriptionGroupConfig(brokerAddrs.iterator().next(), name);
+ if (config == null) {
+ throw new BusinessException(404, "Consumer group not
found: " + name);
+ }
+ return
ConsumerGroupSettingsVO.builder().groupName(name).retryQueueNums(config.getRetryQueueNums())
+ .retryMaxTimes(config.getRetryMaxTimes()).build();
+ } catch (BusinessException exception) {
+ throw exception;
+ } catch (Exception exception) {
+ throw classifyBrokerFailure(exception, "get consumer group
settings");
+ }
+ });
+ }
+
+ @Override
+ public ConsumerGroupSettingsVO updateConsumerGroupSettings(String
instanceId, String name, int retryQueueNums,
+ int
retryMaxTimes) {
+ return executeForInstance(instanceId, admin -> {
+ try {
+ String clusterName = getClusterName(admin);
+ Set<String> brokerAddrs =
getMasterBrokerAddrsForCluster(admin, clusterName);
+ if (brokerAddrs.isEmpty()) {
+ throw new BusinessException(502, "No broker available to
update consumer group settings");
+ }
+ for (String brokerAddr : brokerAddrs) {
+ SubscriptionGroupConfig config =
admin.examineSubscriptionGroupConfig(brokerAddr, name);
+ if (config == null) {
+ throw new BusinessException(404, "Consumer group not
found: " + name);
+ }
+ config.setRetryQueueNums(retryQueueNums);
+ config.setRetryMaxTimes(retryMaxTimes);
+ admin.createAndUpdateSubscriptionGroupConfig(brokerAddr,
config);
+ }
+ RmqGroup group = groupMapper.selectOne(new
LambdaQueryWrapper<RmqGroup>()
+ .eq(RmqGroup::getClusterId, clusterName)
+ .eq(RmqGroup::getInstanceId, metadataScope(instanceId))
+ .eq(RmqGroup::getName, name));
+ if (group != null) {
+ group.setMaxRetry(retryMaxTimes);
+ group.setGmtModified(LocalDateTime.now());
+ groupMapper.updateById(group);
+ }
+ recordAudit("UPDATE_GROUP_SETTINGS", name, "retryQueueNums=" +
retryQueueNums
+ + ", retryMaxTimes=" + retryMaxTimes, "SUCCESS");
+ return
ConsumerGroupSettingsVO.builder().groupName(name).retryQueueNums(retryQueueNums)
+ .retryMaxTimes(retryMaxTimes).build();
+ } catch (BusinessException exception) {
+ recordAudit("UPDATE_GROUP_SETTINGS", name,
exception.getMessage(), "FAILED");
+ throw exception;
+ } catch (Exception exception) {
+ recordAudit("UPDATE_GROUP_SETTINGS", name,
exception.getMessage(), "FAILED");
+ throw classifyBrokerFailure(exception, "update consumer group
settings");
+ }
+ });
+ }
+
private ConsumerGroupVO createConsumerGroup(MQAdminExt admin,
ConsumerGroupVO group) {
String groupName = group.getName();
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 205a6d7cc..a7a260f57 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
@@ -167,6 +167,35 @@ class ConsumerGroupControllerTest {
verifyNoInteractions(metadataService);
}
+ @Test
+ void consumerGroupSettingsShouldUseTheSelectedInstance() throws Exception {
+ ConsumerGroupSettingsVO settings =
ConsumerGroupSettingsVO.builder().groupName("cg-orders")
+ .retryQueueNums(2).retryMaxTimes(8).build();
+ when(metadataService.getConsumerGroupSettings("instance-a",
"cg-orders")).thenReturn(settings);
+
+
mockMvc.perform(get("/api/groups/cg-orders/settings").param("instanceId",
"instance-a"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data.retryQueueNums").value(2));
+
+ verify(metadataService).getConsumerGroupSettings("instance-a",
"cg-orders");
+ }
+
+ @Test
+ void consumerGroupSettingsUpdateShouldValidateAndDelegate() throws
Exception {
+ Map<String, Object> body = Map.of("instanceId", "instance-a", "name",
"cg-orders",
+ "retryQueueNums", 2, "retryMaxTimes", 8);
+ when(metadataService.updateConsumerGroupSettings("instance-a",
"cg-orders", 2, 8))
+
.thenReturn(ConsumerGroupSettingsVO.builder().groupName("cg-orders").retryQueueNums(2)
+ .retryMaxTimes(8).build());
+
+
mockMvc.perform(post("/api/groups/settings").contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(body)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data.retryMaxTimes").value(8));
+
+ verify(metadataService).updateConsumerGroupSettings("instance-a",
"cg-orders", 2, 8);
+ }
+
@Test
void getConsumerStackShouldReturnStackTrace() throws Exception {
ConsumerThreadStackVO thread = ConsumerThreadStackVO.builder()
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 b19ee8a1e..fba406cac 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
@@ -27,11 +27,13 @@ import
org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
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.topic.TopicVO;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
@@ -576,6 +578,43 @@ class RocketMQAdminClientImplTest {
assertThat(captor.getValue().getSqlSegment()).contains("cluster_id",
"instance_id", "name");
}
+ @Test
+ void updateConsumerGroupSettingsPreservesBrokerConfiguration() throws
Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqGroup.class);
+ DefaultMQAdminExt selectedAdmin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ ClusterInfo clusterInfo = new ClusterInfo();
+ clusterInfo.setClusterAddrTable(new HashMap<>(Map.of("cluster-1", new
HashSet<>(List.of("broker-1")))));
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("broker-1");
+ brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
+ clusterInfo.setBrokerAddrTable(new HashMap<>(Map.of("broker-1",
brokerData)));
+ SubscriptionGroupConfig config = new SubscriptionGroupConfig();
+ config.setGroupName("cg-orders");
+ config.setConsumeEnable(false);
+ config.setRetryQueueNums(1);
+ config.setRetryMaxTimes(16);
+ when(selectedAdmin.examineBrokerClusterInfo()).thenReturn(clusterInfo);
+ when(selectedAdmin.examineSubscriptionGroupConfig("10.0.0.1:10911",
"cg-orders")).thenReturn(config);
+ when(groupMapper.selectOne(any())).thenReturn(null);
+
doNothing().when(selectedAdmin).createAndUpdateSubscriptionGroupConfig(anyString(),
any());
+
when(runtimeAdminClientResolver.execute(org.mockito.ArgumentMatchers.eq("instance-a"),
any()))
+ .thenAnswer(invocation -> {
+ MqAdminExtFactory.AdminAction<?> action =
invocation.getArgument(1);
+ return action.apply(selectedAdmin);
+ });
+
+ ConsumerGroupSettingsVO settings =
adminClient.updateConsumerGroupSettings("instance-a", "cg-orders", 2, 8);
+
+ assertThat(settings.getRetryQueueNums()).isEqualTo(2);
+ assertThat(settings.getRetryMaxTimes()).isEqualTo(8);
+ ArgumentCaptor<SubscriptionGroupConfig> captor =
ArgumentCaptor.forClass(SubscriptionGroupConfig.class);
+ verify(selectedAdmin).createAndUpdateSubscriptionGroupConfig(
+ org.mockito.ArgumentMatchers.eq("10.0.0.1:10911"),
captor.capture());
+ assertThat(captor.getValue().isConsumeEnable()).isFalse();
+ assertThat(captor.getValue().getRetryQueueNums()).isEqualTo(2);
+ assertThat(captor.getValue().getRetryMaxTimes()).isEqualTo(8);
+ }
+
@Test
void deleteConsumerGroupUsesSelectedInstanceAdmin() throws Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqGroup.class);
diff --git a/web/src/api/consumerGroups.test.ts
b/web/src/api/consumerGroups.test.ts
index 8ad59ece1..a25044ea1 100644
--- a/web/src/api/consumerGroups.test.ts
+++ b/web/src/api/consumerGroups.test.ts
@@ -20,12 +20,14 @@ import { afterEach, beforeEach, describe, expect, it, vi }
from 'vitest';
import client from './client';
import {
getConsumerGroup,
+ getConsumerGroupSettings,
getConsumerProgress,
getConsumerSubscriptions,
listConsumerGroupPage,
listConsumerGroups,
deleteConsumerGroup,
resetConsumerOffset,
+ updateConsumerGroupSettings,
} from './metadata';
const mock = new MockAdapter(client);
@@ -124,4 +126,26 @@ describe('consumer groups API contract', () => {
await expect(deleteConsumerGroup(group.name,
'instance-1')).resolves.toBeUndefined();
});
+
+ it('gets and updates settings in the selected Apache instance', async () => {
+ const settings = { groupName: group.name, retryQueueNums: 2,
retryMaxTimes: 8 };
+ mock
+ .onGet(`/groups/${encodeURIComponent(group.name)}/settings`, {
+ params: { instanceId: 'instance-1' },
+ })
+ .reply(200, { code: 200, data: settings });
+ await expect(getConsumerGroupSettings(group.name,
'instance-1')).resolves.toEqual(settings);
+
+ const payload = {
+ instanceId: 'instance-1',
+ name: group.name,
+ retryQueueNums: 2,
+ retryMaxTimes: 8,
+ };
+ mock.onPost('/groups/settings').reply((config) => {
+ expect(JSON.parse(config.data)).toEqual(payload);
+ return [200, { code: 200, data: settings }];
+ });
+ await
expect(updateConsumerGroupSettings(payload)).resolves.toEqual(settings);
+ });
});
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 003fc52cb..662ea21ba 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -111,6 +111,12 @@ export interface ConsumerGroupDetail extends ConsumerGroup
{
instances: ConsumerInstance[];
}
+export interface ConsumerGroupSettings {
+ groupName: string;
+ retryQueueNums: number;
+ retryMaxTimes: number;
+}
+
export interface QueueProgress {
topic: string;
broker: string;
@@ -274,6 +280,21 @@ export async function createConsumerGroup(data:
Partial<ConsumerGroup>) {
return res.data.data;
}
+export async function getConsumerGroupSettings(name: string, instanceId:
string) {
+ const res = await client.get<{ data: ConsumerGroupSettings }>(
+ `/groups/${encodeURIComponent(name)}/settings`,
+ { params: { instanceId } },
+ );
+ return res.data.data;
+}
+
+export async function updateConsumerGroupSettings(
+ data: Omit<ConsumerGroupSettings, 'groupName'> & { instanceId: string; name:
string },
+) {
+ const res = await client.post<{ data: ConsumerGroupSettings
}>('/groups/settings', data);
+ return res.data.data;
+}
+
export async function deleteConsumerGroup(name: string, instanceId?: string) {
await client.post('/groups/delete', { name, ...(instanceId ? { instanceId }
: {}) });
}
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index ea364b57c..6001176a1 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -54,6 +54,7 @@ import {
ListBullets,
Info,
ArrowsClockwise,
+ SlidersHorizontal,
} from '@phosphor-icons/react';
import { ImportOutlined, ExportOutlined, DeleteOutlined, SyncOutlined } from
'@ant-design/icons';
import type { ColumnsType } from 'antd/es/table';
@@ -81,6 +82,8 @@ import {
getConsumerSubscriptions,
listConsumerGroupPage,
resetConsumerOffset,
+ getConsumerGroupSettings,
+ updateConsumerGroupSettings,
} from '../../services/consumerService';
import { useInstanceFilter } from '../../hooks/useInstanceFilter';
import {
@@ -193,6 +196,10 @@ const ConsumerPageContent = ({
const [sortKey, setSortKey] = useState<string>('name_asc');
const [modalOpen, setModalOpen] = useState(false);
const [selectedGroup, setSelectedGroup] = useState<ConsumerGroup |
null>(null);
+ const [settingsGroup, setSettingsGroup] = useState<ConsumerGroup |
null>(null);
+ const [settingsLoading, setSettingsLoading] = useState(false);
+ const [settingsSubmitting, setSettingsSubmitting] = useState(false);
+ const [settingsForm] = Form.useForm<{ retryQueueNums: number; retryMaxTimes:
number }>();
const [createModalOpen, setCreateModalOpen] = useState(false);
const [form] = Form.useForm();
const [dataTypeValue, setDataTypeValue] = useState<string |
undefined>(undefined);
@@ -350,6 +357,47 @@ const ConsumerPageContent = ({
void loadProgress(group.name);
};
+ const openSettingsModal = async (group: ConsumerGroup) => {
+ if (!selectedInstanceId) return;
+ setSettingsGroup(group);
+ setSettingsLoading(true);
+ try {
+ const settings = await getConsumerGroupSettings(group.name,
selectedInstanceId);
+ settingsForm.setFieldsValue(settings);
+ } catch {
+ setSettingsGroup(null);
+ message.error('加载消费组配置失败，请稍后重试');
+ } finally {
+ setSettingsLoading(false);
+ }
+ };
+
+ const saveSettings = async () => {
+ if (!settingsGroup || !selectedInstanceId) return;
+ const values = await settingsForm.validateFields();
+ setSettingsSubmitting(true);
+ try {
+ const saved = await updateConsumerGroupSettings({
+ instanceId: selectedInstanceId,
+ name: settingsGroup.name,
+ ...values,
+ });
+ setGroups((current) =>
+ current.map((group) =>
+ group.name === settingsGroup.name
+ ? { ...group, retryMaxTimes: saved.retryMaxTimes }
+ : group,
+ ),
+ );
+ message.success('消费组配置已保存');
+ setSettingsGroup(null);
+ } catch {
+ message.error('保存消费组配置失败，请稍后重试');
+ } finally {
+ setSettingsSubmitting(false);
+ }
+ };
+
const selectedDiagnosticKey = selectedGroup
? diagnosticCacheKey(selectedInstanceId, selectedGroup.name)
: '';
@@ -638,6 +686,17 @@ const ConsumerPageContent = ({
width: 210,
render: (_: unknown, record: ConsumerGroup) => (
<Flex gap={6}>
+ <Button
+ size="small"
+ icon={<SlidersHorizontal size={14} />}
+ disabled={isCloudInstance}
+ onClick={(e) => {
+ e.stopPropagation();
+ void openSettingsModal(record);
+ }}
+ >
+ 配置
+ </Button>
<Button
size="small"
icon={<Eye size={14} />}
@@ -1426,6 +1485,37 @@ const ConsumerPageContent = ({
)}
</Modal>
+ <Modal
+ title="消费组配置"
+ open={Boolean(settingsGroup)}
+ onCancel={() => setSettingsGroup(null)}
+ onOk={() => void saveSettings()}
+ confirmLoading={settingsSubmitting}
+ okText="保存"
+ cancelText="取消"
+ destroyOnHidden
+ >
+ <Form form={settingsForm} layout="vertical" style={{ marginTop: 16 }}>
+ <Form.Item label="Group 名称">
+ <Text strong>{settingsGroup?.name}</Text>
+ </Form.Item>
+ <Form.Item
+ label="重试队列数"
+ name="retryQueueNums"
+ rules={[{ required: true, message: '请输入重试队列数' }]}
+ >
+ <InputNumber min={1} max={128} style={{ width: '100%' }}
disabled={settingsLoading} />
+ </Form.Item>
+ <Form.Item
+ label="最大重试次数"
+ name="retryMaxTimes"
+ rules={[{ required: true, message: '请输入最大重试次数' }]}
+ >
+ <InputNumber min={1} max={128} style={{ width: '100%' }}
disabled={settingsLoading} />
+ </Form.Item>
+ </Form>
+ </Modal>
+
{/* ═══════════════════════════════════════════
Consumer Stack Modal
═══════════════════════════════════════════ */}
diff --git a/web/src/services/consumerService.ts
b/web/src/services/consumerService.ts
index 7ef632a68..db00f75a5 100644
--- a/web/src/services/consumerService.ts
+++ b/web/src/services/consumerService.ts
@@ -4,6 +4,7 @@ import type {
ConsumerGroup,
ConsumerGroupPageQuery,
ConsumerGroupQuery,
+ ConsumerGroupSettings,
ConsumerGroupDetail,
ConsumerStackTrace,
PageResult,
@@ -104,6 +105,26 @@ export async function getConsumerGroup(
return normalizeConsumerGroup(await metadataApi.getConsumerGroup(name,
instanceId));
}
+export async function getConsumerGroupSettings(
+ name: string,
+ instanceId: string,
+): Promise<ConsumerGroupSettings> {
+ if (isMockMode()) return { groupName: name, retryQueueNums: 1,
retryMaxTimes: 16 };
+ return metadataApi.getConsumerGroupSettings(name, instanceId);
+}
+
+export async function updateConsumerGroupSettings(
+ data: Omit<ConsumerGroupSettings, 'groupName'> & { instanceId: string; name:
string },
+) {
+ if (isMockMode())
+ return {
+ groupName: data.name,
+ retryQueueNums: data.retryQueueNums,
+ retryMaxTimes: data.retryMaxTimes,
+ };
+ return metadataApi.updateConsumerGroupSettings(data);
+}
+
export async function getConsumerSubscriptions(
name: string,
instanceId?: string,
