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 904de6fc [ISSUE #1579] Validate Tencent Topic queue counts (#1581)
904de6fc is described below
commit 904de6fcaa67572020d3fc85d1a7991bd31e70e3
Author: youngkermit8-coder <[email protected]>
AuthorDate: Tue Aug 11 20:26:30 2026 +0800
[ISSUE #1579] Validate Tencent Topic queue counts (#1581)
* [ISSUE #1579] Validate Tencent Topic queue counts
Signed-off-by: youngkermit8-coder <[email protected]>
* Normalize Tencent Topic update queue counts
Signed-off-by: youngkermit8-coder <[email protected]>
---------
Signed-off-by: youngkermit8-coder <[email protected]>
---
.../provider/tencent/TencentInstanceProvider.java | 19 +++++-
.../tencent/TencentInstanceProviderTest.java | 78 +++++++++++++++++++++-
2 files changed, 95 insertions(+), 2 deletions(-)
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 16a867ea..c0effbd1 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
@@ -66,6 +66,8 @@ public class TencentInstanceProvider implements
InstanceProvider {
static final int MAX_PAGES = 100;
static final int CONSUMER_PAGE_SIZE = 100;
static final int DEFAULT_QUEUE_NUM = 8;
+ static final int MIN_QUEUE_NUM = 3;
+ static final int MAX_QUEUE_NUM = 16;
private static final String NOT_IMPLEMENTED = "Tencent Cloud operation is
not implemented yet";
private final TencentClientFactory clientFactory;
@@ -182,12 +184,18 @@ public class TencentInstanceProvider implements
InstanceProvider {
request.setInstanceId(context.cloudInstanceId());
request.setTopic(topic.getName());
request.setRemark(topic.getRemark());
+ Long requestedQueueNum = null;
if (topic.getWriteQueues() > 0 || topic.getReadQueues() > 0) {
- request.setQueueNum(queueNum(topic));
+ requestedQueueNum = queueNum(topic);
+ request.setQueueNum(requestedQueueNum);
}
clientFactory.call(context.credentialId(), context.regionId(), client
-> client.ModifyTopic(request));
topic.setInstanceId(instanceId);
topic.setPerm(defaultPerm(topic.getPerm()));
+ if (requestedQueueNum != null) {
+ topic.setWriteQueues(requestedQueueNum.intValue());
+ topic.setReadQueues(requestedQueueNum.intValue());
+ }
topic.setUpdatedAt(LocalDateTime.now());
return topic;
}
@@ -376,6 +384,15 @@ public class TencentInstanceProvider implements
InstanceProvider {
if (topic.getWriteQueues() < 0 || topic.getReadQueues() < 0) {
throw new BusinessException(400, "Topic queue number must not be
negative");
}
+ if (topic.getWriteQueues() > 0 && topic.getReadQueues() > 0
+ && topic.getWriteQueues() != topic.getReadQueues()) {
+ throw new BusinessException(400,
+ "Topic write and read queue numbers must match for Tencent
Cloud");
+ }
+ int queueNum = topic.getWriteQueues() > 0 ? topic.getWriteQueues() :
topic.getReadQueues();
+ if (queueNum > 0 && (queueNum < MIN_QUEUE_NUM || queueNum >
MAX_QUEUE_NUM)) {
+ throw new BusinessException(400, "Topic queue number must be
between 3 and 16");
+ }
}
private static Long queueNum(TopicVO topic) {
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 a9ed54d7..ee353771 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
@@ -25,6 +25,7 @@ import com.tencentcloudapi.trocket.v20230308.models.TopicItem;
import com.tencentcloudapi.trocket.v20230308.TrocketClient;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.InstanceRepository;
import org.apache.rocketmq.studio.instance.InstanceVO;
import org.apache.rocketmq.studio.instance.topic.TopicConsumerVO;
@@ -40,9 +41,13 @@ import java.util.List;
import java.util.Optional;
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.anyString;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -74,7 +79,7 @@ class TencentInstanceProviderTest {
.regionId(REGION)
.credentialId(CREDENTIAL_ID)
.build()));
- when(clientFactory.call(anyString(), anyString(),
any())).thenAnswer(invocation -> {
+ lenient().when(clientFactory.call(anyString(), anyString(),
any())).thenAnswer(invocation -> {
TencentClientFactory.TencentCall<Object> action =
invocation.getArgument(2);
return action.execute(client);
});
@@ -128,6 +133,59 @@ class TencentInstanceProviderTest {
assertThat(created.getInstanceId()).isEqualTo(STUDIO_INSTANCE_ID);
}
+ @Test
+ void createTopicShouldAcceptTencentQueueDefaultsAndBoundariesTest() throws
Exception {
+ when(client.CreateTopic(any())).thenReturn(null);
+ for (int queueNum : new int[]{0, 3, 16}) {
+ TopicVO topic = new TopicVO();
+ topic.setName("orders-" + queueNum);
+ topic.setType(TopicType.NORMAL);
+ topic.setWriteQueues(queueNum);
+ provider.createTopic(STUDIO_INSTANCE_ID, topic);
+ }
+
+ ArgumentCaptor<CreateTopicRequest> captor =
ArgumentCaptor.forClass(CreateTopicRequest.class);
+ verify(client, times(3)).CreateTopic(captor.capture());
+ assertThat(captor.getAllValues())
+ .extracting(CreateTopicRequest::getQueueNum)
+ .containsExactly(8L, 3L, 16L);
+ }
+
+ @Test
+ void createTopicShouldRejectQueueCountsOutsideTencentRangeTest() {
+ TopicVO topic = new TopicVO();
+ topic.setName("orders");
+ topic.setType(TopicType.NORMAL);
+ topic.setWriteQueues(2);
+
+ assertThatThrownBy(() -> provider.createTopic(STUDIO_INSTANCE_ID,
topic))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("between 3 and 16");
+
+ topic.setWriteQueues(17);
+ assertThatThrownBy(() -> provider.createTopic(STUDIO_INSTANCE_ID,
topic))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("between 3 and 16");
+ verifyNoInteractions(client);
+ }
+
+ @Test
+ void topicWritesShouldRejectMismatchedQueueCountsTest() {
+ TopicVO topic = new TopicVO();
+ topic.setName("orders");
+ topic.setType(TopicType.NORMAL);
+ topic.setWriteQueues(8);
+ topic.setReadQueues(4);
+
+ assertThatThrownBy(() -> provider.createTopic(STUDIO_INSTANCE_ID,
topic))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("must match for Tencent Cloud");
+ assertThatThrownBy(() -> provider.updateTopic(STUDIO_INSTANCE_ID,
topic))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("must match for Tencent Cloud");
+ verifyNoInteractions(client);
+ }
+
@Test
void updateAndDeleteTopicShouldCallTencentOpenApiTest() throws Exception {
when(client.ModifyTopic(any())).thenReturn(null);
@@ -145,6 +203,24 @@ class TencentInstanceProviderTest {
assertThat(updateCaptor.getValue().getInstanceId()).isEqualTo(CLOUD_INSTANCE_ID);
assertThat(updateCaptor.getValue().getTopic()).isEqualTo("orders");
assertThat(updateCaptor.getValue().getRemark()).isEqualTo("updated");
+ assertThat(updateCaptor.getValue().getQueueNum()).isNull();
+ }
+
+ @Test
+ void updateTopicShouldNormalizeTheSingleTencentQueueCountTest() throws
Exception {
+ when(client.ModifyTopic(any())).thenReturn(null);
+ TopicVO topic = new TopicVO();
+ topic.setName("orders");
+ topic.setType(TopicType.NORMAL);
+ topic.setWriteQueues(12);
+
+ TopicVO updated = provider.updateTopic(STUDIO_INSTANCE_ID, topic);
+
+ ArgumentCaptor<ModifyTopicRequest> captor =
ArgumentCaptor.forClass(ModifyTopicRequest.class);
+ verify(client).ModifyTopic(captor.capture());
+ assertThat(captor.getValue().getQueueNum()).isEqualTo(12L);
+ assertThat(updated.getWriteQueues()).isEqualTo(12);
+ assertThat(updated.getReadQueues()).isEqualTo(12);
}
@Test
