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 75cba9cf8 fix(apache): honour the delivery order type when creating a
group (#4839)
75cba9cf8 is described below
commit 75cba9cf8e764d7fe2b662ca63a2a177b489b04e
Author: Wang1rrr <[email protected]>
AuthorDate: Sat Sep 26 18:45:09 2026 +0800
fix(apache): honour the delivery order type when creating a group (#4839)
The console's create form submits deliveryOrderType (PARTITON_ORDER /
MESSAGES_ORDER) whenever the subscription data type is FIFO. Aliyun maps it
through normalizeDeliveryOrderType and Tencent feeds it to
setConsumeMessageOrderly via isOrderly, but the Apache provider never read
it: createConsumerGroup set only groupName, consumeEnable,
consumeBroadcastEnable, retryQueueNums and retryMaxTimes, so
consumeMessageOrderly kept its false default and an ordered group was
created unordered. The controller echoed the submitted VO back, so the
browser was told the request succeeded, and the CREATE_GROUP audit detail
mentioned only retryMaxTimes - no trace of the dropped value.
The same flag is already an editable group setting (read at
getConsumerGroupSettings, written at updateConsumerGroupSettings), so the
create path now derives it from the requested delivery order type using the
vocabulary TencentInstanceProvider.isOrderly already accepts, and records
the resolved flag in the audit detail. The AI tool path and the CSV
importer go through the same method, so they are covered too.
---
.../provider/apache/RocketMQAdminClientImpl.java | 24 +++++++-
.../apache/RocketMQAdminClientImplTest.java | 64 ++++++++++++++++++++++
2 files changed, 87 insertions(+), 1 deletion(-)
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 72f2b3671..a72f219c1 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
@@ -72,6 +72,7 @@ import java.util.Comparator;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
@@ -783,6 +784,12 @@ public class RocketMQAdminClientImpl implements
AdminClient {
config.setConsumeBroadcastEnable(true);
config.setRetryQueueNums(1);
config.setRetryMaxTimes(group.getRetryMaxTimes() > 0 ?
group.getRetryMaxTimes() : 16);
+ // The create form submits the ordered spelling of the group's
delivery order type,
+ // and consumeMessageOrderly is the very flag the group settings
dialog reads and
+ // writes, so an ordered create request has to reach the broker
instead of being
+ // dropped on the floor.
+ boolean consumeMessageOrderly =
isOrderlyDelivery(group.getDeliveryOrderType());
+ config.setConsumeMessageOrderly(consumeMessageOrderly);
boolean physicalExists = importing && verifyGroupImport(admin,
brokerAddrs, config);
if (!physicalExists) {
@@ -793,7 +800,8 @@ public class RocketMQAdminClientImpl implements AdminClient
{
persistConsumerGroup(group, groupClusterName,
config.getRetryMaxTimes());
recordAudit("CREATE_GROUP", groupName,
- "retryMaxTimes=" + config.getRetryMaxTimes(), "SUCCESS");
+ "retryMaxTimes=" + config.getRetryMaxTimes()
+ + ", consumeMessageOrderly=" +
consumeMessageOrderly, "SUCCESS");
return group;
} catch (BusinessException e) {
recordAudit("CREATE_GROUP", groupName, e.getMessage(), "FAILED");
@@ -804,6 +812,20 @@ public class RocketMQAdminClientImpl implements
AdminClient {
}
}
+ /**
+ * Ordered-delivery spellings accepted across the console and the
providers: the create form
+ * submits PARTITON_ORDER / MESSAGES_ORDER, the CSV importer also allows
PARTITION_ORDER, and
+ * the cloud providers tolerate FIFO and ORDERLY.
TencentInstanceProvider.isOrderly applies the
+ * same vocabulary, so "ordered" keeps meaning the same thing on every
backend.
+ */
+ private static boolean isOrderlyDelivery(String deliveryOrderType) {
+ if (deliveryOrderType == null || deliveryOrderType.isBlank()) {
+ return false;
+ }
+ String normalized = deliveryOrderType.toUpperCase(Locale.ROOT);
+ return normalized.contains("FIFO") || normalized.contains("ORDER");
+ }
+
private void persistConsumerGroup(ConsumerGroupVO group, String
clusterName, int retryMaxTimes) {
// Ownership is determined by the global reservation guard; a
three-column filter must not mask cross-instance conflicts.
RmqGroup entity = groupMapper.selectOne(new
LambdaQueryWrapper<RmqGroup>()
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 6d6431599..5b53ba63a 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
@@ -64,6 +64,7 @@ import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
import org.junit.jupiter.params.provider.ValueSource;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
@@ -1126,6 +1127,69 @@ class RocketMQAdminClientImplTest {
assertThat(captor.getValue().getSqlSegment()).contains("name").doesNotContain("cluster_id",
"instance_id");
}
+ @ParameterizedTest
+ @ValueSource(strings = {"PARTITON_ORDER", "PARTITION_ORDER",
"MESSAGES_ORDER", "FIFO", "Orderly"})
+ void
createConsumerGroupPropagatesAnOrderedDeliveryOrderTypeToTheBrokerTest(String
deliveryOrderType) 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.setCluster("cluster-1");
+ 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)));
+ when(selectedAdmin.examineBrokerClusterInfo()).thenReturn(clusterInfo);
+ when(groupMapper.selectOne(any())).thenReturn(null);
+
doNothing().when(selectedAdmin).createAndUpdateSubscriptionGroupConfig(anyString(),
any());
+ when(runtimeAdminClientResolver.execute(eq("open-source-local"),
any()))
+ .thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1)
+ .apply(selectedAdmin));
+
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("cg-ordered");
+ group.setInstanceId("open-source-local");
+ group.setDeliveryOrderType(deliveryOrderType);
+
+ adminClient.createConsumerGroup(group);
+
+ ArgumentCaptor<SubscriptionGroupConfig> config =
ArgumentCaptor.forClass(SubscriptionGroupConfig.class);
+
verify(selectedAdmin).createAndUpdateSubscriptionGroupConfig(eq("10.0.0.1:10911"),
config.capture());
+ assertThat(config.getValue().isConsumeMessageOrderly()).isTrue();
+ }
+
+ @ParameterizedTest
+ @NullAndEmptySource
+ @ValueSource(strings = {"Concurrently", "concurrently", " "})
+ void
createConsumerGroupKeepsAnUnorderedDeliveryOrderTypeConcurrentTest(String
deliveryOrderType) 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.setCluster("cluster-1");
+ 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)));
+ when(selectedAdmin.examineBrokerClusterInfo()).thenReturn(clusterInfo);
+ when(groupMapper.selectOne(any())).thenReturn(null);
+
doNothing().when(selectedAdmin).createAndUpdateSubscriptionGroupConfig(anyString(),
any());
+ when(runtimeAdminClientResolver.execute(eq("open-source-local"),
any()))
+ .thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1)
+ .apply(selectedAdmin));
+
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("cg-concurrent");
+ group.setInstanceId("open-source-local");
+ group.setDeliveryOrderType(deliveryOrderType);
+
+ adminClient.createConsumerGroup(group);
+
+ ArgumentCaptor<SubscriptionGroupConfig> config =
ArgumentCaptor.forClass(SubscriptionGroupConfig.class);
+
verify(selectedAdmin).createAndUpdateSubscriptionGroupConfig(eq("10.0.0.1:10911"),
config.capture());
+ assertThat(config.getValue().isConsumeMessageOrderly()).isFalse();
+ }
+
@ParameterizedTest
@ValueSource(ints = {0, 8})
void updateConsumerGroupPreservesEachBrokersConfiguration(int
retryMaxTimes) throws Exception {
