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 431d9791e fix(aliyun): keep an ordered consumer group ordered when
creating it (#4816)
431d9791e is described below
commit 431d9791e6678071afe77a754e737d2e2f13f825
Author: Wang1rrr <[email protected]>
AuthorDate: Thu Sep 24 18:18:15 2026 +0800
fix(aliyun): keep an ordered consumer group ordered when creating it (#4816)
fix(aliyun): keep an ordered consumer group ordered when creating it
normalizeDeliveryOrderType only recognised the FIFO/ORDERLY spellings, but
the consumer group form submits the RocketMQ order types PARTITON_ORDER
(partition ordered) and MESSAGES_ORDER (globally ordered), and the CSV
importer also accepts the PARTITION_ORDER spelling. All of them fell through
to the Concurrently default, so an ordered group created on an Aliyun
instance was submitted as a concurrent one - carrying DefaultRetryPolicy,
which ordered groups reject - and nothing reported the downgrade.
Classify on the same two substrings TencentInstanceProvider#isOrderly
already keys off, so the two adapters reading the same ConsumerGroupVO
value cannot disagree about whether the group is ordered.
---
.../provider/alibaba/AliyunInstanceProvider.java | 14 ++++--
.../alibaba/AliyunInstanceProviderTest.java | 51 ++++++++++++++++++++++
2 files changed, 62 insertions(+), 3 deletions(-)
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 c82d21027..da5a502ff 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
@@ -376,14 +376,22 @@ public class AliyunInstanceProvider implements
InstanceProvider {
}
/**
- * OpenAPI accepts Concurrently/Orderly; tolerate FIFO/ordered spellings
from the UI.
+ * OpenAPI accepts Concurrently/Orderly; tolerate the order-type spellings
the rest of the
+ * console produces. The consumer group form offers {@code PARTITON_ORDER}
(partition ordered)
+ * and {@code MESSAGES_ORDER} (globally ordered), the CSV importer also
accepts the
+ * {@code PARTITION_ORDER} spelling, and older callers used {@code
FIFO}/{@code ORDERLY}.
+ *
+ * <p>The classification deliberately matches {@code
TencentInstanceProvider#isOrderly}, which
+ * keys off the same two substrings: both adapters read the same
+ * {@link ConsumerGroupVO#getDeliveryOrderType()} value and must not
disagree about whether the
+ * group is ordered.
*/
static String normalizeDeliveryOrderType(String raw) {
if (raw == null || raw.isBlank()) {
return DEFAULT_DELIVERY_ORDER_TYPE;
}
- String value = raw.trim();
- if ("FIFO".equalsIgnoreCase(value) ||
"ORDERLY".equalsIgnoreCase(value)) {
+ String value = raw.trim().toUpperCase(Locale.ROOT);
+ if (value.contains("FIFO") || value.contains("ORDER")) {
return ORDERLY_DELIVERY_ORDER_TYPE;
}
return DEFAULT_DELIVERY_ORDER_TYPE;
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 b58cebd0c..2edbd0654 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
@@ -646,6 +646,36 @@ class AliyunInstanceProviderTest {
assertThat(created.getRetryMaxTimes()).isEqualTo(16);
}
+ @Test
+ void createConsumerGroupShouldKeepPartitionOrderedGroupsOrderlyTest() {
+ 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_ordered");
+ // exactly what the console form submits when the subscription data
type is FIFO
+ group.setDeliveryOrderType("PARTITON_ORDER");
+ group.setRetryMaxTimes(5);
+
+ ConsumerGroupVO created =
provider.createConsumerGroup(STUDIO_INSTANCE_ID, group);
+
+ ArgumentCaptor<CreateConsumerGroupRequest> captor =
+ ArgumentCaptor.forClass(CreateConsumerGroupRequest.class);
+ verify(asyncClient).createConsumerGroup(captor.capture());
+ CreateConsumerGroupRequest request = captor.getValue();
+ assertThat(request.getDeliveryOrderType()).isEqualTo("Orderly");
+ // ordered groups reject DefaultRetryPolicy, so the retry policy has
to travel with the type
+
assertThat(request.getConsumeRetryPolicy().getRetryPolicy()).isEqualTo("FixedRetryPolicy");
+
assertThat(request.getConsumeRetryPolicy().getFixedIntervalRetryTime()).isEqualTo(10);
+
assertThat(request.getConsumeRetryPolicy().getMaxRetryTimes()).isEqualTo(5);
+ assertThat(created.getDeliveryOrderType()).isEqualTo("Orderly");
+ }
+
@Test
void resetOffsetShouldUseSpecifiedTimeTest() {
stubInstance();
@@ -1038,6 +1068,27 @@ class AliyunInstanceProviderTest {
AliyunInstanceProvider.normalizeDeliveryOrderType("Concurrently"));
}
+ /**
+ * The consumer group form never submits FIFO/ORDERLY. It submits the
RocketMQ order-type
+ * spellings - PARTITON_ORDER for partition ordered and MESSAGES_ORDER for
globally ordered -
+ * and the CSV importer additionally accepts PARTITION_ORDER. An
unrecognised spelling used to
+ * fall through to Concurrently, so an ordered group created on an Aliyun
instance came back
+ * concurrent with no error anywhere.
+ */
+ @Test
+ void normalizeDeliveryOrderTypeShouldMapConsoleOrderTypesToOrderlyTest() {
+ for (String value : List.of("PARTITON_ORDER", "MESSAGES_ORDER",
"PARTITION_ORDER",
+ "partiton_order", " PARTITON_ORDER ")) {
+
assertThat(AliyunInstanceProvider.normalizeDeliveryOrderType(value))
+ .as("value " + value)
+ .isEqualTo("Orderly");
+ }
+
assertThat(AliyunInstanceProvider.normalizeDeliveryOrderType("Concurrently"))
+ .isEqualTo("Concurrently");
+ assertThat(AliyunInstanceProvider.normalizeDeliveryOrderType(" "))
+ .isEqualTo("Concurrently");
+ }
+
@Test
void timeConversionUsesAliyunUtc8Zone() {
// "2024-01-01 00:00:00" is 2023-12-31T16:00:00Z in UTC+8 regardless
of server zone.
