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 684915725 fix(instance): lifecycle cleanup and credential-scoped
client release (#2233)
684915725 is described below
commit 6849157252f04ce9965bb81aee9c39cb39e26ed5
Author: shown <[email protected]>
AuthorDate: Wed Aug 19 11:20:38 2026 +0800
fix(instance): lifecycle cleanup and credential-scoped client release
(#2233)
* fix(instance): release credential-scoped admin clients and remove metrics
bindings on deletion
When an instance's endpoint or admin credential reference changes, only
stop the
cached admin client that is actually unreferenced: release every identity
on the
endpoint only when no other instance uses it, otherwise evict the single
credential-scoped client so other identities sharing the endpoint keep
running.
Deleting an instance now also strips its identifier from every metrics data
source binding inside the same transaction, so dashboards no longer
reference a
removed instance.
Signed-off-by: yuluo-yx <[email protected]>
* fix(credentials): prevent deleted credential recreation on update
Credential updates used save(), which re-inserts the row when it vanished
between lookup and persistence, silently resurrecting a concurrently deleted
credential. Switch the update path to a replace() that performs a pure
update
and reports whether a row changed, returning 404 instead of recreating the
credential.
Signed-off-by: yuluo-yx <[email protected]>
---------
Signed-off-by: yuluo-yx <[email protected]>
---
.../studio/cluster/broker/MqAdminExtFactory.java | 19 +++++++
.../rocketmq/studio/instance/InstanceService.java | 58 ++++++++++++++++------
.../credential/CloudCredentialRepository.java | 2 +
.../credential/CloudCredentialService.java | 12 +++--
.../MybatisPlusCloudCredentialRepository.java | 5 ++
.../cluster/broker/MqAdminExtFactoryTest.java | 16 ++++++
.../studio/instance/InstanceServiceTest.java | 25 ++++++++++
.../credential/CloudCredentialServiceTest.java | 22 +++++++-
8 files changed, 137 insertions(+), 22 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
index 0377f8671..b491b8636 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
@@ -132,6 +132,25 @@ public class MqAdminExtFactory {
log.info("Released RocketMQ admin clients for namesrv {}",
normalizedNamesrvAddr);
}
+ /** Stops one credential-scoped client without interrupting other
identities on the endpoint. */
+ public void release(String namesrvAddr, String authenticationIdentity) {
+ if (namesrvAddr == null || namesrvAddr.isBlank()) {
+ return;
+ }
+ String normalizedNamesrvAddr = normalizeNamesrvAddr(namesrvAddr);
+ if (normalizedNamesrvAddr.isEmpty()) {
+ return;
+ }
+ AdminClientCacheKey key = new
AdminClientCacheKey(normalizedNamesrvAddr,
+ normalizeAuthenticationIdentity(authenticationIdentity));
+ DefaultMQAdminExt admin = cache.remove(key);
+ if (admin != null) {
+ safeShutdown(admin);
+ log.info("Released RocketMQ admin client for namesrv {} and
identity {}",
+ normalizedNamesrvAddr, key.authenticationIdentity());
+ }
+ }
+
private DefaultMQAdminExt createAndStart(String namesrvAddr, RPCHook
rpcHook) {
DefaultMQAdminExt admin = newAdmin(rpcHook);
admin.setNamesrvAddr(namesrvAddr);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
index e3d6002d9..00c0db05a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
@@ -29,9 +29,12 @@ import
org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.provider.CloudInstanceDetailVO;
import org.apache.rocketmq.studio.provider.InstanceProvider;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
+import org.apache.rocketmq.studio.settings.DataSourceVO;
+import org.apache.rocketmq.studio.settings.SettingsRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.util.ArrayList;
@@ -50,6 +53,7 @@ public class InstanceService {
private final InstanceProviderRegistry providerRegistry;
private final MqAdminExtFactory adminFactory;
private final OperationAuditService operationAuditService;
+ private final SettingsRepository settingsRepository;
public List<InstanceVO> listInstances(InstanceType type, String search) {
log.debug("Listing instances, type={}, search={}", type, search);
@@ -292,6 +296,7 @@ public class InstanceService {
return saved;
}
+ @Transactional
public void deleteInstance(Long id) {
log.info("Deleting instance: {}", id);
@@ -314,11 +319,28 @@ public class InstanceService {
if (!instanceRepository.deleteById(id)) {
throw new BusinessException(404, "InstanceVO not found: " + id);
}
+ removeDataSourceBindings(existing.getName());
releaseApacheEndpointIfUnused(existing, null);
recordAudit("DELETE_INSTANCE", "INSTANCE", String.valueOf(id), null,
instanceAuditDetail(existing));
}
+ private void removeDataSourceBindings(String instanceId) {
+ for (DataSourceVO dataSource :
settingsRepository.findAllDataSources()) {
+ List<String> instanceIds = dataSource.getInstanceIds();
+ if (instanceIds == null || !instanceIds.contains(instanceId)) {
+ continue;
+ }
+ dataSource.setInstanceIds(instanceIds.stream()
+ .filter(candidate -> !instanceId.equals(candidate))
+ .toList());
+ if (!settingsRepository.replaceDataSource(dataSource)) {
+ log.warn("Metrics data source {} disappeared while removing
instance binding {}",
+ dataSource.getKey(), instanceId);
+ }
+ }
+ }
+
private void requireInstance(InstanceVO instance) {
if (instance == null) {
throw new BusinessException(400, "Instance request is required");
@@ -355,7 +377,7 @@ public class InstanceService {
if (vendor != InstanceVendor.APACHE) {
return;
}
- releaseEndpointIfUnused(existing.getEndpoint(), currentEndpoint,
existing.getId());
+ releaseOldClientIfUnused(existing, currentEndpoint, null,
existing.getId());
}
private void releaseApacheClientIfChanged(InstanceVO existing, InstanceVO
saved) {
@@ -363,25 +385,31 @@ public class InstanceService {
if (vendor != InstanceVendor.APACHE) {
return;
}
- if
(!Objects.equals(normalizeCredentialRef(existing.getAdminCredentialRef()),
- normalizeCredentialRef(saved.getAdminCredentialRef()))
- && Objects.equals(normalizeEndpoint(existing.getEndpoint()),
normalizeEndpoint(saved.getEndpoint()))) {
- adminFactory.release(existing.getEndpoint());
- return;
- }
- releaseEndpointIfUnused(existing.getEndpoint(), saved.getEndpoint(),
existing.getId());
+ releaseOldClientIfUnused(existing, saved.getEndpoint(),
saved.getAdminCredentialRef(), existing.getId());
}
- private void releaseEndpointIfUnused(String previousEndpoint, String
currentEndpoint, Long excludedInstanceId) {
- String oldEndpoint = normalizeEndpoint(previousEndpoint);
- if (oldEndpoint == null ||
oldEndpoint.equals(normalizeEndpoint(currentEndpoint))) {
+ private void releaseOldClientIfUnused(InstanceVO existing, String
currentEndpoint,
+ String currentCredentialRef, Long
excludedInstanceId) {
+ String oldEndpoint = normalizeEndpoint(existing.getEndpoint());
+ String oldCredentialRef =
normalizeCredentialRef(existing.getAdminCredentialRef());
+ if (oldEndpoint == null ||
oldEndpoint.equals(normalizeEndpoint(currentEndpoint))
+ && Objects.equals(oldCredentialRef,
normalizeCredentialRef(currentCredentialRef))) {
return;
}
- boolean stillReferenced = instanceRepository.findAll().stream()
- .anyMatch(instance ->
!excludedInstanceId.equals(instance.getId())
- &&
oldEndpoint.equals(normalizeEndpoint(instance.getEndpoint())));
- if (!stillReferenced) {
+ List<InstanceVO> remaining = instanceRepository.findAll().stream()
+ .filter(instance ->
!excludedInstanceId.equals(instance.getId()))
+ .toList();
+ boolean endpointReferenced = remaining.stream()
+ .anyMatch(instance ->
oldEndpoint.equals(normalizeEndpoint(instance.getEndpoint())));
+ if (!endpointReferenced) {
adminFactory.release(oldEndpoint);
+ return;
+ }
+ boolean identityReferenced = remaining.stream()
+ .anyMatch(instance ->
oldEndpoint.equals(normalizeEndpoint(instance.getEndpoint()))
+ && Objects.equals(oldCredentialRef,
normalizeCredentialRef(instance.getAdminCredentialRef())));
+ if (!identityReferenced) {
+ adminFactory.release(oldEndpoint, oldCredentialRef);
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialRepository.java
index 91aff5a28..9e76d9f7c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialRepository.java
@@ -31,5 +31,7 @@ public interface CloudCredentialRepository {
CloudCredentialVO save(CloudCredentialVO credential);
+ boolean replace(CloudCredentialVO credential);
+
boolean deleteById(Long id);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialService.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialService.java
index 5dd010fa1..b0ad47bde 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialService.java
@@ -97,11 +97,13 @@ public class CloudCredentialService {
existing.setRemark(request.getRemark());
}
existing.setGmtModified(LocalDateTime.now());
- CloudCredentialVO saved = credentialRepository.save(existing);
- invalidateCloudClients(saved);
- recordAudit("UPDATE_CLOUD_CREDENTIAL", "CLOUD_CREDENTIAL",
String.valueOf(saved.getId()), null,
- credentialAuditDetail(saved));
- return maskAccessKey(saved);
+ if (!credentialRepository.replace(existing)) {
+ throw new BusinessException(404, "Cloud credential not found: " +
request.getId());
+ }
+ invalidateCloudClients(existing);
+ recordAudit("UPDATE_CLOUD_CREDENTIAL", "CLOUD_CREDENTIAL",
String.valueOf(existing.getId()), null,
+ credentialAuditDetail(existing));
+ return maskAccessKey(existing);
}
public void delete(Long id) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepository.java
index 262f5b36d..36e555e49 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepository.java
@@ -82,6 +82,11 @@ public class MybatisPlusCloudCredentialRepository implements
CloudCredentialRepo
return credential;
}
+ @Override
+ public boolean replace(CloudCredentialVO credential) {
+ return credential.getId() != null &&
credentialMapper.updateById(toEntity(credential)) > 0;
+ }
+
@Override
public boolean deleteById(Long id) {
return id != null && credentialMapper.deleteById(id) > 0;
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
index d04e0f7d7..b7ee85ad0 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
@@ -84,6 +84,22 @@ class MqAdminExtFactoryTest {
verify(admin).shutdown();
}
+ @Test
+ void releaseShouldEvictOnlyMatchingCredentialIdentityTest() throws
Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RecordingFactory factory = new RecordingFactory(admin);
+
+ factory.execute("10.0.0.1:9876", null, "credential-a", ignored ->
null);
+ factory.execute("10.0.0.1:9876", null, "credential-b", ignored ->
null);
+ factory.release("10.0.0.1:9876", "credential-a");
+ factory.execute("10.0.0.1:9876", null, "credential-b", ignored ->
null);
+ factory.execute("10.0.0.1:9876", null, "credential-a", ignored ->
null);
+
+ assertThat(factory.created.get()).isEqualTo(3);
+ verify(admin, times(3)).start();
+ verify(admin).shutdown();
+ }
+
@Test
void releaseShouldEvictClientUsingEquivalentNameServerAddressList() throws
Exception {
DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
index 2e6f93ce4..5ce91ec65 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
@@ -28,6 +28,8 @@ import
org.apache.rocketmq.studio.provider.CloudCatalogProvider;
import org.apache.rocketmq.studio.provider.CloudInstanceDetailVO;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.apache.rocketmq.studio.provider.InstanceProvider;
+import org.apache.rocketmq.studio.settings.DataSourceVO;
+import org.apache.rocketmq.studio.settings.SettingsRepository;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -71,6 +73,9 @@ class InstanceServiceTest {
@Mock
private OperationAuditService operationAuditService;
+ @Mock
+ private SettingsRepository settingsRepository;
+
@InjectMocks
private InstanceService instanceService;
@@ -744,6 +749,26 @@ class InstanceServiceTest {
verify(adminFactory).release("namesrv:9876");
}
+ @Test
+ void deleteInstanceShouldRemoveMetricsDataSourceBindingTest() {
+ InstanceVO existing = InstanceVO.builder().name("to-delete").build();
+ existing.setId(1L);
+ DataSourceVO dataSource = DataSourceVO.builder().key("prometheus")
+ .instanceIds(List.of("to-delete", "inst-2")).build();
+
when(instanceRepository.findById(1L)).thenReturn(Optional.of(existing));
+
when(providerRegistry.forVendor(InstanceVendor.APACHE)).thenReturn(instanceProvider);
+ when(instanceProvider.countTopics("1")).thenReturn(0);
+ when(instanceProvider.countGroups("1")).thenReturn(0);
+ when(instanceRepository.deleteById(1L)).thenReturn(true);
+
when(settingsRepository.findAllDataSources()).thenReturn(List.of(dataSource));
+
when(settingsRepository.replaceDataSource(dataSource)).thenReturn(true);
+
+ instanceService.deleteInstance(1L);
+
+ assertThat(dataSource.getInstanceIds()).containsExactly("inst-2");
+ verify(settingsRepository).replaceDataSource(dataSource);
+ }
+
@Test
void deleteInstanceShouldRejectConcurrentRemoval() {
InstanceVO existing = InstanceVO.builder()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialServiceTest.java
index ddb5d4005..89d248f98 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialServiceTest.java
@@ -167,8 +167,7 @@ class CloudCredentialServiceTest {
stored.setAccessKey("LTAI5tUpdateKey000000001");
stored.setSecretKey("old-secret");
when(credentialRepository.findById(1L)).thenReturn(Optional.of(stored));
- when(credentialRepository.save(any(CloudCredentialVO.class)))
- .thenAnswer(invocation -> invocation.getArgument(0));
+
when(credentialRepository.replace(any(CloudCredentialVO.class))).thenReturn(true);
UpdateCloudCredentialDTO request = new UpdateCloudCredentialDTO();
request.setId(1L);
@@ -181,6 +180,25 @@ class CloudCredentialServiceTest {
eq("1"), eq(null), eq("name=null, vendor=ALIYUN"),
eq("SUCCESS"), eq(null));
}
+ @Test
+ void updateShouldNotRecreateConcurrentlyDeletedCredentialTest() {
+ CloudCredentialVO stored = new CloudCredentialVO();
+ stored.setId(1L);
+ stored.setVendor(InstanceVendor.TENCENT);
+ stored.setAccessKey("AKIDexample");
+
when(credentialRepository.findById(1L)).thenReturn(Optional.of(stored));
+ when(credentialRepository.replace(stored)).thenReturn(false);
+ UpdateCloudCredentialDTO request = new UpdateCloudCredentialDTO();
+ request.setId(1L);
+ request.setName("renamed");
+
+ assertThatThrownBy(() -> service.update(request))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Cloud credential not found: 1");
+
+ verify(tencentClientFactory, never()).invalidateCredential(any());
+ }
+
@Test
void deleteShouldInvalidateAliyunClientsAfterRemovingCredentialTest() {
CloudCredentialVO stored = new CloudCredentialVO();
