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 05682d759 fix(group): apply or reject the consume model of a cloud
consumer group (#6548)
05682d759 is described below
commit 05682d759ebea1ec25f7ced457b3a1e8cd5c613c
Author: Apulupie <[email protected]>
AuthorDate: Thu Oct 15 18:15:17 2026 +0800
fix(group): apply or reject the consume model of a cloud consumer group
(#6548)
The group-create contract carries ConsumerGroupVO.consumeType, and the tool
and REST responses
echo the submitted VO, but neither cloud adapter applied the requested
model.
AliyunInstanceProvider.createConsumerGroup never set messageModel even
though the OpenAPI
request has the field, so a broadcast request created a clustering group;
the group page reads
the real model back through
AliyunConverters.toConsumeType(data.getMessageModel()), so the
downgrade surfaced later as an unexplained change. The request now carries
the vendor's own
spelling for a requested broadcast, while an unspecified or clustering
model keeps the field out
of the payload exactly as before.
TencentInstanceProvider.createConsumerGroup cannot apply it at all:
CreateConsumerGroupRequest
has no consume-model field (only instanceId, consumerGroup, maxRetryTimes,
consumeEnable,
consumeMessageOrderly, remark, tagList, retryPolicy, liteTopic), while
enrichConsumerGroupDetail reports DescribeConsumerGroup's real
consumeModel. validateConsumerGroup
now rejects a broadcast request with 400 before any SDK call, which is what
InstanceProvider#updateConsumerGroup's contract asks for - never report a
change the adapter
cannot apply - and both adapters still apply the fields their APIs do
accept (maxRetryTimes,
consumeMessageOrderly, deliveryOrderType).
---
.../studio/provider/alibaba/AliyunConverters.java | 10 +++++
.../provider/alibaba/AliyunInstanceProvider.java | 13 ++++--
.../provider/tencent/TencentInstanceProvider.java | 9 ++++
.../alibaba/AliyunInstanceProviderTest.java | 50 ++++++++++++++++++++++
.../tencent/TencentInstanceProviderTest.java | 33 ++++++++++++++
5 files changed, 112 insertions(+), 3 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
index 708bea0b9..a5eded738 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
@@ -196,6 +196,16 @@ final class AliyunConverters {
return ConsumeType.CLUSTERING;
}
+ /**
+ * The inverse of {@link #toConsumeType}: the OpenAPI's own spelling for a
requested consume model
+ * ({@code Broadcasting} / {@code Clustering} — the values the read paths
recognise). Null for an
+ * unspecified or clustering model, so a request that does not ask for
broadcast keeps the field
+ * out of the payload and the vendor default in charge.
+ */
+ static String toMessageModel(ConsumeType consumeType) {
+ return consumeType == ConsumeType.BROADCASTING ? "Broadcasting" : null;
+ }
+
static List<QueueProgressVO>
toQueueProgressRows(GetConsumerGroupLagResponseBody.Data data) {
List<QueueProgressVO> rows = new ArrayList<>();
Map<String, DataTopicLagMapValue> topicLagMap = data.getTopicLagMap();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
index a23f302b9..58db22f71 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
@@ -382,12 +382,19 @@ public class AliyunInstanceProvider implements
InstanceProvider {
} else {
retryPolicy.retryPolicy(DEFAULT_RETRY_POLICY);
}
- CreateConsumerGroupRequest request =
CreateConsumerGroupRequest.builder()
+ CreateConsumerGroupRequest.Builder requestBuilder =
CreateConsumerGroupRequest.builder()
.instanceId(ctx.cloudInstanceId())
.consumerGroupId(group.getName())
.deliveryOrderType(deliveryOrderType)
- .consumeRetryPolicy(retryPolicy.build())
- .build();
+ .consumeRetryPolicy(retryPolicy.build());
+ // A requested broadcast model used to be dropped here while the
response echoed it back and
+ // the group page read the vendor's real (clustering) model afterwards
- the same silent
+ // downgrade #4816 fixed for the delivery order type.
+ String messageModel =
AliyunConverters.toMessageModel(group.getConsumeType());
+ if (messageModel != null) {
+ requestBuilder.messageModel(messageModel);
+ }
+ CreateConsumerGroupRequest request = requestBuilder.build();
clientFactory.call(ctx.credentialId(), ctx.regionId(), client ->
client.createConsumerGroup(request));
group.setInstanceId(instanceId);
group.setDeliveryOrderType(deliveryOrderType);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
index 4d2628c8c..fcd93cd3b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
@@ -1083,6 +1083,15 @@ public class TencentInstanceProvider implements
InstanceProvider {
if (group == null || !StringUtils.hasText(group.getName())) {
throw new BusinessException(400, "Consumer group name is
required");
}
+ // CreateConsumerGroupRequest has no consume-model field: the OpenAPI
always creates a
+ // clustering group, so a requested broadcast model cannot be applied.
Answer that instead of
+ // echoing the request back and letting DescribeConsumerGroup report
the vendor's real model
+ // later - InstanceProvider#updateConsumerGroup's contract asks a
provider never to report a
+ // change it cannot apply.
+ if (group.getConsumeType() == ConsumeType.BROADCASTING) {
+ throw new BusinessException(400, "Tencent consumer groups always
consume in clustering mode; "
+ + "consumeType BROADCASTING cannot be applied");
+ }
}
private static void requireGroupName(String groupName) {
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 3e9dcad1d..88b3551d9 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
@@ -814,6 +814,56 @@ class AliyunInstanceProviderTest {
assertThat(created.getDeliveryOrderType()).isEqualTo("Orderly");
}
+ @Test
+ void createConsumerGroupShouldApplyTheRequestedBroadcastingModelTest() {
+ stubInstance();
+ stubCallThrough();
+ when(asyncClient.createConsumerGroup(any()))
+
.thenReturn(CompletableFuture.completedFuture(CreateConsumerGroupResponse.create()
+ .toBuilder()
+ .statusCode(200)
+
.body(CreateConsumerGroupResponseBody.builder().data(true).build())
+ .build()));
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("GID_broadcast");
+ // what the AI tool's group-create input and any API caller can send
+ group.setConsumeType(ConsumeType.BROADCASTING);
+ group.setRetryMaxTimes(5);
+
+ ConsumerGroupVO created =
provider.createConsumerGroup(STUDIO_INSTANCE_ID, group);
+
+ ArgumentCaptor<CreateConsumerGroupRequest> captor =
+ ArgumentCaptor.forClass(CreateConsumerGroupRequest.class);
+ verify(asyncClient).createConsumerGroup(captor.capture());
+ // The vendor's spelling, the one the group page reads back through
+ // AliyunConverters.toConsumeType(data.getMessageModel()).
+
assertThat(captor.getValue().getMessageModel()).isEqualTo("Broadcasting");
+
assertThat(created.getConsumeType()).isEqualTo(ConsumeType.BROADCASTING);
+ }
+
+ @Test
+ void
createConsumerGroupShouldLeaveTheConsumeModelToTheVendorWhenClusteringTest() {
+ stubInstance();
+ stubCallThrough();
+ when(asyncClient.createConsumerGroup(any()))
+
.thenReturn(CompletableFuture.completedFuture(CreateConsumerGroupResponse.create()
+ .toBuilder()
+ .statusCode(200)
+
.body(CreateConsumerGroupResponseBody.builder().data(true).build())
+ .build()));
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("GID_clustered");
+ group.setConsumeType(ConsumeType.CLUSTERING);
+
+ provider.createConsumerGroup(STUDIO_INSTANCE_ID, group);
+
+ ArgumentCaptor<CreateConsumerGroupRequest> captor =
+ ArgumentCaptor.forClass(CreateConsumerGroupRequest.class);
+ verify(asyncClient).createConsumerGroup(captor.capture());
+ // Only a requested broadcast is sent: the clustering default keeps
the request byte-identical.
+ assertThat(captor.getValue().getMessageModel()).isNull();
+ }
+
@Test
void resetOffsetShouldUseSpecifiedTimeTest() {
stubInstance();
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 7710327c7..fb22003aa 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
@@ -712,6 +712,39 @@ class TencentInstanceProviderTest {
assertThat(created.getRetryMaxTimes()).isEqualTo(20);
}
+ @Test
+ void
createConsumerGroupShouldRejectABroadcastingModelTheOpenApiCannotApplyTest()
throws Exception {
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("GID_broadcast");
+ // what the AI tool's group-create input and any API caller can send
+ group.setConsumeType(ConsumeType.BROADCASTING);
+
+ assertThatThrownBy(() ->
provider.createConsumerGroup(STUDIO_INSTANCE_ID, group))
+ .isInstanceOfSatisfying(BusinessException.class, error -> {
+ assertThat(error.getCode()).isEqualTo(400);
+ assertThat(error.getMessage()).contains("clustering");
+ });
+ // CreateConsumerGroupRequest has no consume-model field, so nothing
may be sent.
+ verifyNoInteractions(client);
+ }
+
+ @Test
+ void createConsumerGroupShouldStillApplyTheOrderedAndRetrySettingsTest()
throws Exception {
+ when(client.CreateConsumerGroup(any())).thenReturn(null);
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("GID_ordered");
+ group.setDeliveryOrderType("MESSAGES_ORDER");
+ group.setRetryMaxTimes(20);
+
+ provider.createConsumerGroup(STUDIO_INSTANCE_ID, group);
+
+ ArgumentCaptor<CreateConsumerGroupRequest> captor =
ArgumentCaptor.forClass(CreateConsumerGroupRequest.class);
+ verify(client).CreateConsumerGroup(captor.capture());
+ assertThat(captor.getValue().getConsumeMessageOrderly()).isTrue();
+ assertThat(captor.getValue().getMaxRetryTimes()).isEqualTo(20L);
+ assertThat(captor.getValue().getConsumeEnable()).isTrue();
+ }
+
@Test
void deleteConsumerGroupShouldCallTencentOpenApiTest() throws Exception {
when(client.DeleteConsumerGroup(any())).thenReturn(null);
