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 99765228 fix: query producer connections with producer group (#999)
99765228 is described below
commit 99765228e6500423ab95a1a1d2aad54fd3acee93
Author: yx9o <[email protected]>
AuthorDate: Wed Aug 5 17:42:27 2026 +0800
fix: query producer connections with producer group (#999)
---
.../studio/cluster/client/ClientProvider.java | 2 +
.../studio/cluster/client/ClientProviderStub.java | 7 +
.../cluster/client/ProducerConnectionService.java | 28 ++--
.../studio/cluster/client/ProducerController.java | 9 ++
.../studio/rocketmq/RocketMQClientProvider.java | 180 ++++++++++++++-------
.../cluster/client/ClientProviderStubTest.java | 8 +
.../client/ProducerConnectionServiceTest.java | 85 +++-------
.../cluster/client/ProducerControllerTest.java | 34 ++++
.../rocketmq/RocketMQClientProviderTest.java | 121 ++++++++++++--
web/src/api/producer.test.ts | 11 --
web/src/api/producer.ts | 7 +-
web/src/pages/studio/Producer.tsx | 8 +-
web/src/pages/studio/__tests__/Producer.test.tsx | 21 ++-
13 files changed, 347 insertions(+), 174 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProvider.java
index 543d0caf..86a58c50 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProvider.java
@@ -21,4 +21,6 @@ import java.util.List;
public interface ClientProvider {
List<ClientConnectionVO> findConnections(String clusterId, String type);
+
+ List<ClientConnectionVO> findProducerConnections(String topic, String
producerGroup);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStub.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStub.java
index 8ca03c03..c5bb198a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStub.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStub.java
@@ -30,4 +30,11 @@ public class ClientProviderStub implements ClientProvider {
clusterId, type);
throw new BusinessException(501, "Client connection provider is not
configured");
}
+
+ @Override
+ public List<ClientConnectionVO> findProducerConnections(String topic,
String producerGroup) {
+ log.warn("ClientProviderStub.findProducerConnections called without a
real client provider. "
+ + "topic={}, producerGroup={}", topic, producerGroup);
+ throw new BusinessException(501, "Client connection provider is not
configured");
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionService.java
index bce19e91..646bc4fb 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionService.java
@@ -16,7 +16,7 @@
*/
package org.apache.rocketmq.studio.cluster.client;
-import org.apache.rocketmq.studio.common.domain.enums.ClientType;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
@@ -28,28 +28,17 @@ import java.util.List;
@RequiredArgsConstructor
public class ProducerConnectionService {
- private final ClientService clientService;
+ private final ClientProvider clientProvider;
public List<ProducerConnectionVO> listConnections(String topic, String
producerGroup) {
log.info("Listing producer connections, topic={}, producerGroup={}",
topic, producerGroup);
- String normalizedTopic = normalizeFilter(topic);
- String normalizedProducerGroup = normalizeFilter(producerGroup);
- return clientService.listConnections(null,
ClientType.Producer.name()).stream()
- .filter(connection -> matchesFilter(connection,
normalizedTopic, normalizedProducerGroup))
+ String normalizedTopic = requireFilter(topic, "topic");
+ String normalizedProducerGroup = requireFilter(producerGroup,
"producerGroup");
+ return clientProvider.findProducerConnections(normalizedTopic,
normalizedProducerGroup).stream()
.map(this::toProducerConnection)
.toList();
}
- private boolean matchesFilter(ClientConnectionVO connection, String topic,
String producerGroup) {
- if (hasText(topic) && !topic.equals(connection.getGroupOrTopic())) {
- return false;
- }
- if (hasText(producerGroup) &&
!producerGroup.equals(connection.getProducerGroup())) {
- return false;
- }
- return true;
- }
-
private ProducerConnectionVO toProducerConnection(ClientConnectionVO
connection) {
return ProducerConnectionVO.builder()
.clientId(connection.getClientId())
@@ -63,7 +52,10 @@ public class ProducerConnectionService {
return value != null && !value.trim().isEmpty();
}
- private String normalizeFilter(String value) {
- return hasText(value) ? value.trim() : null;
+ private String requireFilter(String value, String fieldName) {
+ if (!hasText(value)) {
+ throw new BusinessException(400, fieldName + " is required");
+ }
+ return value.trim();
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerController.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerController.java
index 1d83ad1a..9967f57d 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerController.java
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.studio.cluster.client;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
@@ -33,6 +34,14 @@ public class ProducerController {
public ProducerConnectionResultVO listConnections(
@RequestParam(required = false) String topic,
@RequestParam(required = false) String producerGroup) {
+ requireParameter(topic, "topic");
+ requireParameter(producerGroup, "producerGroup");
return new
ProducerConnectionResultVO(producerConnectionService.listConnections(topic,
producerGroup));
}
+
+ private void requireParameter(String value, String name) {
+ if (value == null || value.isBlank()) {
+ throw new BusinessException(400, name + " is required");
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
index 157e6794..1683f332 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
@@ -20,9 +20,10 @@ import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerInfo;
import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo;
import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.client.ClientConnectionVO;
import org.apache.rocketmq.studio.cluster.client.ClientProvider;
@@ -37,14 +38,17 @@ import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
import java.util.Set;
/**
* Live {@link ClientProvider} backed by the RocketMQ admin API. It discovers
producer
- * connections by scanning non-system topics and consumer connections by
scanning
+ * connections from each broker's producer table and consumer connections by
scanning
* subscription groups across all brokers in the cluster.
*/
@Slf4j
@@ -52,12 +56,6 @@ import java.util.Set;
@Primary
public class RocketMQClientProvider implements ClientProvider {
- /**
- * Upper bound on the number of non-system topics scanned for producer
connections,
- * to avoid issuing an admin call per topic on clusters with a large topic
count.
- */
- private static final int MAX_PRODUCER_TOPIC_SCAN = 50;
-
private static final long SUBSCRIPTION_GROUP_TIMEOUT_MILLIS = 5000L;
private final ObjectProvider<DefaultMQAdminExt> adminExtProvider;
@@ -68,16 +66,12 @@ public class RocketMQClientProvider implements
ClientProvider {
@Override
public List<ClientConnectionVO> findConnections(String clusterId, String
type) {
- DefaultMQAdminExt adminExt = adminExtProvider.getIfAvailable();
- if (adminExt == null) {
- log.warn("DefaultMQAdminExt is not configured, returning empty
client connection list");
- return List.of();
- }
+ DefaultMQAdminExt adminExt = requireAdminExt();
ClientType clientType = parseType(type);
List<ClientConnectionVO> connections = new ArrayList<>();
if (clientType == null || clientType == ClientType.Producer) {
- connections.addAll(findProducerConnections(adminExt, clusterId));
+ connections.addAll(findAllProducerConnections(adminExt,
clusterId));
}
if (clientType == null || clientType == ClientType.Consumer) {
connections.addAll(findConsumerConnections(adminExt, clusterId));
@@ -85,45 +79,106 @@ public class RocketMQClientProvider implements
ClientProvider {
return connections;
}
- private List<ClientConnectionVO> findProducerConnections(DefaultMQAdminExt
adminExt, String clusterId) {
- List<ClientConnectionVO> result = new ArrayList<>();
- Set<String> topics;
+ @Override
+ public List<ClientConnectionVO> findProducerConnections(String topic,
String producerGroup) {
+ DefaultMQAdminExt adminExt = requireAdminExt();
try {
- TopicList topicList = adminExt.fetchAllTopicList();
- topics = topicList == null ? Set.of() : topicList.getTopicList();
+ ProducerConnection producerConnection =
+ adminExt.examineProducerConnectionInfo(producerGroup,
topic);
+ if (producerConnection == null ||
producerConnection.getConnectionSet() == null) {
+ return List.of();
+ }
+ return producerConnection.getConnectionSet().stream()
+ .filter(Objects::nonNull)
+ .map(connection -> toConnectionVO(
+ connection, ClientType.Producer, topic,
producerGroup, null))
+ .toList();
} catch (Exception e) {
- log.warn("Failed to fetch topic list for producer connection
scan", e);
- return result;
+ throw new BusinessException(502,
+ "Failed to query producer connections: " + rootMessage(e));
}
+ }
- int scanned = 0;
- boolean capped = false;
- for (String topic : topics.stream().filter(topic ->
!isSystemTopic(topic)).sorted().toList()) {
- if (scanned >= MAX_PRODUCER_TOPIC_SCAN) {
- log.warn("Producer connection scan capped at {} non-system
topics", MAX_PRODUCER_TOPIC_SCAN);
- capped = true;
- break;
- }
- scanned++;
+ private List<ClientConnectionVO> findAllProducerConnections(
+ DefaultMQAdminExt adminExt, String clusterId) {
+ Set<String> brokerAddresses = collectProducerBrokerAddresses(adminExt);
+ Map<String, ClientConnectionVO> connections = new LinkedHashMap<>();
+ int successfulBrokers = 0;
+ for (String brokerAddress : brokerAddresses) {
try {
- ProducerConnection producerConnection =
adminExt.examineProducerConnectionInfo(null, topic);
- if (producerConnection == null ||
producerConnection.getConnectionSet() == null) {
- continue;
- }
- for (Connection connection :
producerConnection.getConnectionSet()) {
- if (connection == null) {
- continue;
- }
- result.add(toConnectionVO(connection, ClientType.Producer,
topic, topic, clusterId));
- }
+ ProducerTableInfo producerTable =
adminExt.getAllProducerInfo(brokerAddress);
+ successfulBrokers++;
+ addProducerConnections(connections, producerTable, clusterId);
} catch (Exception e) {
- log.warn("Failed to examine producer connection for topic={},
skipping", topic, e);
+ log.warn("Failed to fetch producer connections from broker={},
skipping", brokerAddress, e);
}
}
- if (capped) {
- result.forEach(connection -> connection.setPartial(true));
+ if (!brokerAddresses.isEmpty() && successfulBrokers == 0) {
+ throw new BusinessException(502, "Failed to query producer
connections from all brokers");
}
- return result;
+ return new ArrayList<>(connections.values());
+ }
+
+ private Set<String> collectProducerBrokerAddresses(DefaultMQAdminExt
adminExt) {
+ ClusterInfo clusterInfo;
+ try {
+ clusterInfo = adminExt.examineBrokerClusterInfo();
+ } catch (Exception e) {
+ throw new BusinessException(502,
+ "Failed to discover brokers for producer connections: " +
rootMessage(e));
+ }
+ Set<String> addresses = new LinkedHashSet<>();
+ if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
+ return addresses;
+ }
+ for (BrokerData brokerData :
clusterInfo.getBrokerAddrTable().values()) {
+ if (brokerData == null) {
+ continue;
+ }
+ String brokerAddress = brokerData.selectBrokerAddr();
+ if (brokerAddress != null && !brokerAddress.isBlank()) {
+ addresses.add(brokerAddress);
+ }
+ }
+ return addresses;
+ }
+
+ private void addProducerConnections(
+ Map<String, ClientConnectionVO> connections,
+ ProducerTableInfo producerTable,
+ String clusterId) {
+ if (producerTable == null || producerTable.getData() == null) {
+ return;
+ }
+ producerTable.getData().forEach((producerGroup, producerInfos) -> {
+ if (producerGroup == null || producerGroup.isBlank() ||
producerInfos == null) {
+ return;
+ }
+ for (ProducerInfo producerInfo : producerInfos) {
+ if (producerInfo == null) {
+ continue;
+ }
+ String key = producerGroup + '\0'
+ + Objects.toString(producerInfo.getClientId(), "") +
'\0'
+ + Objects.toString(producerInfo.getRemoteIP(), "");
+ connections.putIfAbsent(key, toConnectionVO(producerInfo,
producerGroup, clusterId));
+ }
+ });
+ }
+
+ private ClientConnectionVO toConnectionVO(
+ ProducerInfo producerInfo, String producerGroup, String clusterId)
{
+ return ClientConnectionVO.builder()
+ .clientId(producerInfo.getClientId())
+ .type(ClientType.Producer)
+ .groupOrTopic(producerGroup)
+ .producerGroup(producerGroup)
+ .protocol(Protocol.Remoting)
+ .address(producerInfo.getRemoteIP())
+ .language(mapLanguage(producerInfo.getLanguage()))
+ .version(String.valueOf(producerInfo.getVersion()))
+ .clusterName(clusterId)
+ .build();
}
private List<ClientConnectionVO> findConsumerConnections(DefaultMQAdminExt
adminExt, String clusterId) {
@@ -241,24 +296,6 @@ public class RocketMQClientProvider implements
ClientProvider {
}
}
- private boolean isSystemTopic(String topic) {
- if (topic == null) {
- return true;
- }
- return topic.startsWith("RMQ_SYS_")
- || topic.startsWith("SCHEDULE_TOPIC_")
- || topic.startsWith("%RETRY%")
- || topic.startsWith("%DLQ%")
- || topic.startsWith("TBW102")
- || topic.startsWith("SELF_TEST_")
- || topic.startsWith("DefaultCluster")
- || topic.startsWith("broker_")
- || topic.startsWith("OFFSET_MOVED_")
- || topic.startsWith("CID_RMQ_SYS_")
- || topic.startsWith("TRANS_CHECK_")
- || topic.startsWith("BenchmarkTest");
- }
-
private boolean isSystemGroup(String group) {
if (group == null) {
return true;
@@ -272,4 +309,23 @@ public class RocketMQClientProvider implements
ClientProvider {
|| group.startsWith("SELF_TEST_")
|| group.startsWith("CID_HOUSEKEEPING");
}
+
+ private DefaultMQAdminExt requireAdminExt() {
+ DefaultMQAdminExt adminExt = adminExtProvider.getIfAvailable();
+ if (adminExt == null) {
+ throw new BusinessException(501, "Client connection provider is
not configured");
+ }
+ return adminExt;
+ }
+
+ private String rootMessage(Throwable error) {
+ Throwable current = error;
+ while (current.getCause() != null) {
+ current = current.getCause();
+ }
+ String message = current.getMessage();
+ return message == null || message.isBlank()
+ ? current.getClass().getSimpleName()
+ : message;
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStubTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStubTest.java
index 6ee00471..bfefbe90 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStubTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStubTest.java
@@ -33,6 +33,14 @@ class ClientProviderStubTest {
.satisfies(ex -> assertThatBusinessExceptionCode(ex, 501));
}
+ @Test
+ void findProducerConnectionsShouldFailWhenRealProviderIsMissing() {
+ assertThatThrownBy(() ->
provider.findProducerConnections("order-topic", "order-producer"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Client connection provider is not configured")
+ .satisfies(ex -> assertThatBusinessExceptionCode(ex, 501));
+ }
+
private void assertThatBusinessExceptionCode(Throwable ex, int code) {
org.assertj.core.api.Assertions.assertThat(((BusinessException)
ex).getCode()).isEqualTo(code);
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionServiceTest.java
index 489f43cc..7bdbb473 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionServiceTest.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.cluster.client;
import org.apache.rocketmq.studio.common.domain.enums.ClientLanguage;
import org.apache.rocketmq.studio.common.domain.enums.ClientType;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -27,6 +28,8 @@ import org.mockito.junit.jupiter.MockitoExtension;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -34,13 +37,13 @@ import static org.mockito.Mockito.when;
class ProducerConnectionServiceTest {
@Mock
- private ClientService clientService;
+ private ClientProvider clientProvider;
@InjectMocks
private ProducerConnectionService producerConnectionService;
@Test
- void listConnectionsShouldProjectProducerClientsByTopic() {
+ void listConnectionsShouldQueryAndProjectExactProducerGroup() {
ClientConnectionVO producer = ClientConnectionVO.builder()
.clientId("producer-1")
.type(ClientType.Producer)
@@ -50,17 +53,8 @@ class ProducerConnectionServiceTest {
.language(ClientLanguage.Java)
.version("5.1.0")
.build();
- ClientConnectionVO otherProducer = ClientConnectionVO.builder()
- .clientId("producer-2")
- .type(ClientType.Producer)
- .groupOrTopic("payment-topic")
- .producerGroup("pg-payment")
- .address("10.0.0.2:38888")
- .language(ClientLanguage.Go)
- .version("5.0.0")
- .build();
- when(clientService.listConnections(null, ClientType.Producer.name()))
- .thenReturn(List.of(producer, otherProducer));
+ when(clientProvider.findProducerConnections("order-topic", "pg-order"))
+ .thenReturn(List.of(producer));
List<ProducerConnectionVO> result =
producerConnectionService.listConnections("order-topic", "pg-order");
@@ -69,62 +63,35 @@ class ProducerConnectionServiceTest {
assertThat(result.get(0).getClientAddr()).isEqualTo("10.0.0.1:38888");
assertThat(result.get(0).getLanguage()).isEqualTo("Java");
assertThat(result.get(0).getVersionDesc()).isEqualTo("5.1.0");
- verify(clientService).listConnections(null,
ClientType.Producer.name());
+ verify(clientProvider).findProducerConnections("order-topic",
"pg-order");
}
@Test
- void listConnectionsShouldFallbackToProducerGroupWhenTopicIsMissing() {
- ClientConnectionVO producer = ClientConnectionVO.builder()
- .clientId("producer-1")
- .type(ClientType.Producer)
- .groupOrTopic("order-topic")
- .producerGroup("pg-order")
- .address("10.0.0.1:38888")
- .language(ClientLanguage.Java)
- .version("5.1.0")
- .build();
- when(clientService.listConnections(null,
ClientType.Producer.name())).thenReturn(List.of(producer));
-
- List<ProducerConnectionVO> result =
producerConnectionService.listConnections(null, "pg-order");
-
- assertThat(result).hasSize(1);
- assertThat(result.get(0).getClientId()).isEqualTo("producer-1");
+ void listConnectionsShouldRejectMissingTopic() {
+ assertThatThrownBy(() -> producerConnectionService.listConnections("
", "pg-order"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("topic is required")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+ verifyNoInteractions(clientProvider);
}
@Test
- void listConnectionsShouldTrimFilterValues() {
- ClientConnectionVO producer = ClientConnectionVO.builder()
- .clientId("producer-1")
- .type(ClientType.Producer)
- .groupOrTopic("order-topic")
- .producerGroup("pg-order")
- .address("10.0.0.1:38888")
- .language(ClientLanguage.Java)
- .version("5.1.0")
- .build();
- when(clientService.listConnections(null,
ClientType.Producer.name())).thenReturn(List.of(producer));
-
- List<ProducerConnectionVO> result =
producerConnectionService.listConnections(" order-topic ", " pg-order ");
-
- assertThat(result).hasSize(1);
- assertThat(result.get(0).getClientId()).isEqualTo("producer-1");
+ void listConnectionsShouldRejectMissingProducerGroup() {
+ assertThatThrownBy(() ->
producerConnectionService.listConnections("order-topic", null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("producerGroup is required")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+ verifyNoInteractions(clientProvider);
}
@Test
- void listConnectionsShouldRequireProducerGroupWhenBothFiltersAreProvided()
{
- ClientConnectionVO producer = ClientConnectionVO.builder()
- .clientId("producer-1")
- .type(ClientType.Producer)
- .groupOrTopic("order-topic")
- .producerGroup("pg-order")
- .address("10.0.0.1:38888")
- .language(ClientLanguage.Java)
- .version("5.1.0")
- .build();
- when(clientService.listConnections(null,
ClientType.Producer.name())).thenReturn(List.of(producer));
-
- List<ProducerConnectionVO> result =
producerConnectionService.listConnections("order-topic", "wrong-group");
+ void listConnectionsShouldTrimRequiredValues() {
+ when(clientProvider.findProducerConnections("order-topic", "pg-order"))
+ .thenReturn(List.of());
+ List<ProducerConnectionVO> result =
producerConnectionService.listConnections(
+ " order-topic ", " pg-order ");
assertThat(result).isEmpty();
+ verify(clientProvider).findProducerConnections("order-topic",
"pg-order");
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerControllerTest.java
index 6559ffbc..0341a9ec 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerControllerTest.java
@@ -26,6 +26,7 @@ import org.springframework.test.web.servlet.MockMvc;
import java.util.List;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
import static
org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
@@ -64,4 +65,37 @@ class ProducerControllerTest {
verify(producerConnectionService).listConnections("order-topic",
"pg-order");
}
+
+ @Test
+ void listConnectionsShouldRequireTopic() throws Exception {
+ mockMvc.perform(get("/api/producer/connection")
+ .param("producerGroup", "pg-order"))
+ .andExpect(status().isBadRequest())
+ .andExpect(jsonPath("$.code").value(400))
+ .andExpect(jsonPath("$.message").value("topic is required"));
+
+ verifyNoInteractions(producerConnectionService);
+ }
+
+ @Test
+ void listConnectionsShouldRequireProducerGroup() throws Exception {
+ mockMvc.perform(get("/api/producer/connection")
+ .param("topic", "order-topic"))
+ .andExpect(status().isBadRequest())
+ .andExpect(jsonPath("$.code").value(400))
+ .andExpect(jsonPath("$.message").value("producerGroup is
required"));
+
+ verifyNoInteractions(producerConnectionService);
+ }
+
+ @Test
+ void listConnectionsShouldRejectBlankParameters() throws Exception {
+ mockMvc.perform(get("/api/producer/connection")
+ .param("topic", " ")
+ .param("producerGroup", "pg-order"))
+ .andExpect(status().isBadRequest())
+ .andExpect(jsonPath("$.message").value("topic is required"));
+
+ verifyNoInteractions(producerConnectionService);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProviderTest.java
index e3acd9dd..9efcc907 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProviderTest.java
@@ -20,12 +20,14 @@ import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerInfo;
import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo;
import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.studio.cluster.client.ClientConnectionVO;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -41,6 +43,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -64,32 +67,102 @@ class RocketMQClientProviderTest {
}
@Test
- void producerScanTreatsNullTopicSetAsEmpty() throws Exception {
- when(adminExt.fetchAllTopicList()).thenReturn(new TopicList());
+ void producerScanTreatsMissingBrokerMetadataAsEmpty() throws Exception {
+ ClusterInfo clusterInfo = new ClusterInfo();
+ clusterInfo.setBrokerAddrTable(null);
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo);
List<ClientConnectionVO> connections =
provider.findConnections("cluster-a", "Producer");
assertThat(connections).isEmpty();
- verify(adminExt).fetchAllTopicList();
+ verify(adminExt).examineBrokerClusterInfo();
+ verify(adminExt, never()).getAllProducerInfo(anyString());
verify(adminExt, never()).examineProducerConnectionInfo(anyString(),
anyString());
}
@Test
- void producerScanSkipsNullConnectionEntries() throws Exception {
- TopicList topicList = new TopicList();
- topicList.setTopicList(new HashSet<>(List.of("TopicA")));
- ProducerConnection producerConnection = new ProducerConnection();
- HashSet<Connection> connectionSet = new HashSet<>();
- connectionSet.add(null);
- connectionSet.add(connection("producer-client", "10.0.0.1:1000"));
- producerConnection.setConnectionSet(connectionSet);
- when(adminExt.fetchAllTopicList()).thenReturn(topicList);
- when(adminExt.examineProducerConnectionInfo(null,
"TopicA")).thenReturn(producerConnection);
+ void producerScanAggregatesAndDeduplicatesBrokerProducerTables() throws
Exception {
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo(
+ "127.0.0.1:10911", "127.0.0.2:10911"));
+ ProducerInfo shared = producerInfo("producer-client", "10.0.0.1:1000");
+ ProducerInfo another = producerInfo("producer-client-2",
"10.0.0.2:1000");
+ when(adminExt.getAllProducerInfo("127.0.0.1:10911"))
+ .thenReturn(new ProducerTableInfo(Map.of("pg-order",
List.of(shared))));
+ when(adminExt.getAllProducerInfo("127.0.0.2:10911"))
+ .thenReturn(new ProducerTableInfo(Map.of(
+ "pg-order", List.of(shared),
+ "pg-payment", List.of(another))));
List<ClientConnectionVO> connections =
provider.findConnections("cluster-a", "Producer");
- assertThat(connections).hasSize(1);
-
assertThat(connections.get(0).getClientId()).isEqualTo("producer-client");
+ assertThat(connections).hasSize(2);
+ assertThat(connections)
+ .extracting(ClientConnectionVO::getProducerGroup)
+ .containsExactlyInAnyOrder("pg-order", "pg-payment");
+ assertThat(connections)
+ .extracting(ClientConnectionVO::getGroupOrTopic)
+ .containsExactlyInAnyOrder("pg-order", "pg-payment");
+ verify(adminExt, never()).examineProducerConnectionInfo(anyString(),
anyString());
+ }
+
+ @Test
+ void producerScanReturnsPartialResultsWhenOneBrokerFails() throws
Exception {
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo(
+ "127.0.0.1:10911", "127.0.0.2:10911"));
+ when(adminExt.getAllProducerInfo("127.0.0.1:10911"))
+ .thenThrow(new IllegalStateException("broker unavailable"));
+ when(adminExt.getAllProducerInfo("127.0.0.2:10911"))
+ .thenReturn(new ProducerTableInfo(Map.of(
+ "pg-order", List.of(producerInfo("producer-client",
"10.0.0.1:1000")))));
+
+ List<ClientConnectionVO> connections =
provider.findConnections("cluster-a", "Producer");
+
+ assertThat(connections).singleElement().satisfies(connection -> {
+ assertThat(connection.getClientId()).isEqualTo("producer-client");
+ assertThat(connection.getProducerGroup()).isEqualTo("pg-order");
+ });
+ }
+
+ @Test
+ void producerScanFailsWhenEveryBrokerQueryFails() throws Exception {
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo(
+ "127.0.0.1:10911", "127.0.0.2:10911"));
+ when(adminExt.getAllProducerInfo(anyString()))
+ .thenThrow(new IllegalStateException("broker unavailable"));
+
+ assertThatThrownBy(() -> provider.findConnections("cluster-a",
"Producer"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Failed to query producer connections from all
brokers")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(502));
+ }
+
+ @Test
+ void exactProducerQueryPassesNonBlankGroupToAdminApi() throws Exception {
+ ProducerConnection producerConnection = new ProducerConnection();
+ producerConnection.setConnectionSet(new HashSet<>(List.of(
+ connection("producer-client", "10.0.0.1:1000"))));
+ when(adminExt.examineProducerConnectionInfo("pg-order", "TopicA"))
+ .thenReturn(producerConnection);
+
+ List<ClientConnectionVO> connections =
provider.findProducerConnections("TopicA", "pg-order");
+
+ assertThat(connections).singleElement().satisfies(connection -> {
+ assertThat(connection.getClientId()).isEqualTo("producer-client");
+ assertThat(connection.getGroupOrTopic()).isEqualTo("TopicA");
+ assertThat(connection.getProducerGroup()).isEqualTo("pg-order");
+ });
+ verify(adminExt).examineProducerConnectionInfo("pg-order", "TopicA");
+ }
+
+ @Test
+ void exactProducerQueryTranslatesAdminFailureToBadGateway() throws
Exception {
+ when(adminExt.examineProducerConnectionInfo("pg-order", "TopicA"))
+ .thenThrow(new IllegalStateException("broker unavailable"));
+
+ assertThatThrownBy(() -> provider.findProducerConnections("TopicA",
"pg-order"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Failed to query producer connections: broker
unavailable")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(502));
}
@Test
@@ -144,4 +217,20 @@ class RocketMQClientProviderTest {
connection.setVersion(500);
return connection;
}
+
+ private static ProducerInfo producerInfo(String clientId, String remoteIp)
{
+ return new ProducerInfo(clientId, remoteIp, LanguageCode.JAVA, 500,
1000L);
+ }
+
+ private static ClusterInfo clusterInfo(String... brokerAddresses) {
+ ClusterInfo clusterInfo = new ClusterInfo();
+ Map<String, BrokerData> brokerAddrTable = new HashMap<>();
+ for (int i = 0; i < brokerAddresses.length; i++) {
+ String brokerName = "broker-" + i;
+ brokerAddrTable.put(brokerName, new BrokerData(
+ "cluster-a", brokerName, new HashMap<>(Map.of(0L,
brokerAddresses[i]))));
+ }
+ clusterInfo.setBrokerAddrTable(brokerAddrTable);
+ return clusterInfo;
+ }
}
diff --git a/web/src/api/producer.test.ts b/web/src/api/producer.test.ts
index 07bffaea..b5b3f31c 100644
--- a/web/src/api/producer.test.ts
+++ b/web/src/api/producer.test.ts
@@ -85,17 +85,6 @@ describe('Producer API', () => {
expect(result[0].clientId).toBe('producer-1');
});
- it('queries producer connections by topic without a group', async () => {
- mock.onGet('/producer/connection').reply((config) => {
- expect(config.params).toEqual({ topic: 'order-events' });
- expect(config.params).not.toHaveProperty('producerGroup');
- return [200, { connectionSet: [] }];
- });
-
- const result = await queryProducerConnection('order-events');
- expect(result).toEqual([]);
- });
-
it('handles empty producer connections', async () => {
mock.onGet('/producer/connection').reply(200, { connectionSet: [] });
diff --git a/web/src/api/producer.ts b/web/src/api/producer.ts
index c7e93928..89d55fa1 100644
--- a/web/src/api/producer.ts
+++ b/web/src/api/producer.ts
@@ -43,14 +43,13 @@ export async function fetchTopicList(): Promise<string[]> {
return topics.sort();
}
-/** Query producer connections by topic and an optional group */
+/** Query producer connections by topic and producer group */
export async function queryProducerConnection(
topic: string,
- producerGroup?: string,
+ producerGroup: string,
): Promise<ProducerConnection[]> {
- const params = producerGroup ? { topic, producerGroup } : { topic };
const res = await client.get<{ connectionSet: ProducerConnection[]
}>('/producer/connection', {
- params,
+ params: { topic, producerGroup },
});
return res.data?.connectionSet ?? [];
}
diff --git a/web/src/pages/studio/Producer.tsx
b/web/src/pages/studio/Producer.tsx
index 8dba66ea..5e1f1c9c 100644
--- a/web/src/pages/studio/Producer.tsx
+++ b/web/src/pages/studio/Producer.tsx
@@ -57,7 +57,7 @@ const ProducerPage = () => {
};
}, [fetchTopicFailedMessage, message]);
- const onFinish = async (values: { selectedTopic: string; producerGroup?:
string }) => {
+ const onFinish = async (values: { selectedTopic: string; producerGroup:
string }) => {
setLoading(true);
try {
const connections = await queryProducerConnection(values.selectedTopic,
values.producerGroup);
@@ -122,7 +122,11 @@ const ProducerPage = () => {
options={topicList.map((topic) => ({ value: topic, label: topic
}))}
/>
</Form.Item>
- <Form.Item label="PRODUCER GROUP" name="producerGroup">
+ <Form.Item
+ label="PRODUCER GROUP"
+ name="producerGroup"
+ rules={[{ required: true, whitespace: true, message:
t('producer.inputGroup') }]}
+ >
<Input placeholder={t('producer.inputGroup')} style={{ width: 300
}} />
</Form.Item>
<Form.Item>
diff --git a/web/src/pages/studio/__tests__/Producer.test.tsx
b/web/src/pages/studio/__tests__/Producer.test.tsx
index b44ed70f..17e7f241 100644
--- a/web/src/pages/studio/__tests__/Producer.test.tsx
+++ b/web/src/pages/studio/__tests__/Producer.test.tsx
@@ -80,7 +80,7 @@ describe('ProducerPage', () => {
expect(await screen.findByRole('option', { name: 'payment-events'
})).toBeInTheDocument();
});
- it('queries all producer connections for a topic without requiring a group',
async () => {
+ it('queries producer connections with the required topic and group', async
() => {
const user = userEvent.setup();
vi.mocked(queryProducerConnection).mockResolvedValue([
{
@@ -98,11 +98,28 @@ describe('ProducerPage', () => {
await user.click(
await screen.findByText('order-events', { selector:
'.ant-select-item-option-content' }),
);
+ await user.type(screen.getByRole('textbox'), 'order-producer');
await user.click(screen.getByRole('button', { name: /搜索/ }));
await waitFor(() => {
- expect(queryProducerConnection).toHaveBeenCalledWith('order-events',
undefined);
+ expect(queryProducerConnection).toHaveBeenCalledWith('order-events',
'order-producer');
});
expect(await screen.findByText('producer-1')).toBeInTheDocument();
});
+
+ it('does not query without a producer group', async () => {
+ const user = userEvent.setup();
+ renderWithProviders(<ProducerPage />);
+
+ await waitFor(() => expect(fetchTopicList).toHaveBeenCalledTimes(1));
+ const topicSelect = screen.getByRole('combobox');
+ fireEvent.mouseDown(topicSelect.parentElement!);
+ await user.click(
+ await screen.findByText('order-events', { selector:
'.ant-select-item-option-content' }),
+ );
+ await user.click(screen.getByRole('button', { name: /搜索/ }));
+
+ expect(await screen.findByText('请输入生产者组')).toBeInTheDocument();
+ expect(queryProducerConnection).not.toHaveBeenCalled();
+ });
});
