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 4db9a1c4 [ISSUE #1583] Paginate Tencent Topic subscriptions (#1584)
4db9a1c4 is described below
commit 4db9a1c4434a4b375acfbe4ed8105d3a74c8f032
Author: youngkermit8-coder <[email protected]>
AuthorDate: Tue Aug 11 20:27:04 2026 +0800
[ISSUE #1583] Paginate Tencent Topic subscriptions (#1584)
* [ISSUE #1583] Paginate Tencent Topic subscriptions
Signed-off-by: youngkermit8-coder <[email protected]>
* Avoid cross-PR Mockito import conflict
Signed-off-by: youngkermit8-coder <[email protected]>
---------
Signed-off-by: youngkermit8-coder <[email protected]>
---
.../provider/tencent/TencentInstanceProvider.java | 38 +++++++++++++---------
.../tencent/TencentInstanceProviderTest.java | 37 +++++++++++++++++++++
2 files changed, 60 insertions(+), 15 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 c0effbd1..f1d8c6a3 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
@@ -214,23 +214,31 @@ public class TencentInstanceProvider implements
InstanceProvider {
public List<TopicConsumerVO> getTopicConsumers(String instanceId, String
topicName) {
Context context = resolve(instanceId);
requireTopicName(topicName);
- DescribeTopicRequest request = new DescribeTopicRequest();
- request.setInstanceId(context.cloudInstanceId());
- request.setTopic(topicName);
- request.setOffset(0L);
- request.setLimit((long) CONSUMER_PAGE_SIZE);
- DescribeTopicResponse response =
clientFactory.call(context.credentialId(), context.regionId(),
- client -> client.DescribeTopic(request));
- SubscriptionData[] data = response == null ? null :
response.getSubscriptionData();
List<TopicConsumerVO> consumers = new ArrayList<>();
- if (data == null) {
- return consumers;
- }
- for (SubscriptionData subscription : data) {
- if (subscription == null) {
- continue;
+ long fetchedSubscriptions = 0L;
+ for (int page = 0; page < MAX_PAGES; page++) {
+ DescribeTopicRequest request = new DescribeTopicRequest();
+ request.setInstanceId(context.cloudInstanceId());
+ request.setTopic(topicName);
+ request.setOffset((long) page * CONSUMER_PAGE_SIZE);
+ request.setLimit((long) CONSUMER_PAGE_SIZE);
+ DescribeTopicResponse response =
clientFactory.call(context.credentialId(), context.regionId(),
+ client -> client.DescribeTopic(request));
+ SubscriptionData[] data = response == null ? null :
response.getSubscriptionData();
+ if (data == null || data.length == 0) {
+ break;
+ }
+ fetchedSubscriptions += data.length;
+ for (SubscriptionData subscription : data) {
+ if (subscription != null) {
+ consumers.add(toTopicConsumer(subscription));
+ }
+ }
+ Long subscriptionCount = response.getSubscriptionCount();
+ if (data.length < CONSUMER_PAGE_SIZE
+ || subscriptionCount != null && fetchedSubscriptions >=
subscriptionCount) {
+ break;
}
- consumers.add(toTopicConsumer(subscription));
}
return consumers;
}
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 ee353771..de79ba53 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
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.provider.tencent;
import com.tencentcloudapi.trocket.v20230308.models.CreateTopicRequest;
import com.tencentcloudapi.trocket.v20230308.models.DescribeTopicListResponse;
+import com.tencentcloudapi.trocket.v20230308.models.DescribeTopicRequest;
import com.tencentcloudapi.trocket.v20230308.models.DescribeTopicResponse;
import com.tencentcloudapi.trocket.v20230308.models.ModifyTopicRequest;
import com.tencentcloudapi.trocket.v20230308.models.SubscriptionData;
@@ -39,6 +40,7 @@ import org.mockito.junit.jupiter.MockitoExtension;
import java.util.List;
import java.util.Optional;
+import java.util.stream.IntStream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -241,6 +243,33 @@ class TencentInstanceProviderTest {
assertThat(consumers.get(0).getDiffTotal()).isEqualTo(42L);
}
+ @Test
+ void getTopicConsumersShouldReadEverySubscriptionPageTest() throws
Exception {
+ DescribeTopicResponse firstPage = new DescribeTopicResponse();
+ firstPage.setSubscriptionCount(101L);
+ firstPage.setSubscriptionData(IntStream.range(0, 100)
+ .mapToObj(index -> subscription("GID_" + index))
+ .toArray(SubscriptionData[]::new));
+ DescribeTopicResponse secondPage = new DescribeTopicResponse();
+ secondPage.setSubscriptionCount(101L);
+ secondPage.setSubscriptionData(new
SubscriptionData[]{subscription("GID_100")});
+ when(client.DescribeTopic(any())).thenReturn(firstPage, secondPage);
+
+ List<TopicConsumerVO> consumers =
provider.getTopicConsumers(STUDIO_INSTANCE_ID, "orders");
+
+ assertThat(consumers).hasSize(101);
+ assertThat(consumers.get(0).getGroup()).isEqualTo("GID_0");
+ assertThat(consumers.get(100).getGroup()).isEqualTo("GID_100");
+ ArgumentCaptor<DescribeTopicRequest> captor =
ArgumentCaptor.forClass(DescribeTopicRequest.class);
+ verify(client,
org.mockito.Mockito.times(2)).DescribeTopic(captor.capture());
+ assertThat(captor.getAllValues())
+ .extracting(DescribeTopicRequest::getOffset)
+ .containsExactly(0L, 100L);
+ assertThat(captor.getAllValues())
+ .extracting(DescribeTopicRequest::getLimit)
+ .containsOnly(100L);
+ }
+
private static TopicItem topicItem(String name, String type, long
queueNum) {
TopicItem item = new TopicItem();
item.setTopic(name);
@@ -248,4 +277,12 @@ class TencentInstanceProviderTest {
item.setQueueNum(queueNum);
return item;
}
+
+ private static SubscriptionData subscription(String group) {
+ SubscriptionData subscription = new SubscriptionData();
+ subscription.setConsumerGroup(group);
+ subscription.setConsumeType("CLUSTERING");
+ subscription.setMessageModel("CLUSTERING");
+ return subscription;
+ }
}
