This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new adf92d4de8 fix(etcd): prevent watcher leaks in selectInstances (#7435)
adf92d4de8 is described below
commit adf92d4de87c7a05d07bc0e2fd44ba20d468bb94
Author: lymerin <[email protected]>
AuthorDate: Thu Oct 8 14:40:41 2026 +0800
fix(etcd): prevent watcher leaks in selectInstances (#7435)
---
.../etcd/EtcdInstanceRegisterRepository.java | 41 +-------
.../etcd/EtcdInstanceRegisterRepositoryTest.java | 107 ++++++++++++++++++---
2 files changed, 97 insertions(+), 51 deletions(-)
diff --git
a/shenyu-registry/shenyu-registry-etcd/src/main/java/org/apache/shenyu/registry/etcd/EtcdInstanceRegisterRepository.java
b/shenyu-registry/shenyu-registry-etcd/src/main/java/org/apache/shenyu/registry/etcd/EtcdInstanceRegisterRepository.java
index 0acb3f73de..ac6a8656bf 100644
---
a/shenyu-registry/shenyu-registry-etcd/src/main/java/org/apache/shenyu/registry/etcd/EtcdInstanceRegisterRepository.java
+++
b/shenyu-registry/shenyu-registry-etcd/src/main/java/org/apache/shenyu/registry/etcd/EtcdInstanceRegisterRepository.java
@@ -36,13 +36,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.URI;
-import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.function.Function;
import java.util.stream.Collectors;
import static java.nio.charset.StandardCharsets.UTF_8;
@@ -57,8 +54,6 @@ public class EtcdInstanceRegisterRepository implements
ShenyuInstanceRegisterRep
private EtcdClient client;
- private final Map<String, List<InstanceEntity>> watcherInstanceRegisterMap
= new ConcurrentHashMap<>();
-
private final Multimap<String, Watch.Watcher> watchCache =
ArrayListMultimap.create();
@Override
@@ -89,37 +84,11 @@ public class EtcdInstanceRegisterRepository implements
ShenyuInstanceRegisterRep
@Override
public List<InstanceEntity> selectInstances(final String selectKey) {
final String watchKey =
InstancePathConstants.buildInstanceParentPath(selectKey);
- final Function<Map<String, String>, List<InstanceEntity>>
getInstanceRegisterFun = childrenList ->
- childrenList.values().stream().map(x -> {
- InstanceEntity instanceEntity =
GsonUtils.getInstance().fromJson(x, InstanceEntity.class);
- instanceEntity.setUri(getURI(x, instanceEntity.getPort(),
instanceEntity.getHost()));
- return instanceEntity;
- }).collect(Collectors.toList());
- if (Objects.nonNull(watcherInstanceRegisterMap.get(selectKey))) {
- return
getInstanceRegisterFun.apply(client.getKeysMapByPrefix(watchKey));
- }
- Map<String, String> serverNodes = new
ConcurrentHashMap<>(client.getKeysMapByPrefix(watchKey));
- this.client.watchKeyChanges(watchKey, Watch.listener(response -> {
- for (WatchEvent event : response.getEvents()) {
- String value =
event.getKeyValue().getValue().toString(StandardCharsets.UTF_8);
- String path =
event.getKeyValue().getKey().toString(StandardCharsets.UTF_8);
- switch (event.getEventType()) {
- case PUT:
- serverNodes.put(path, value);
- LOGGER.info("watch key {} updated, value is {}",
watchKey, value);
- continue;
- case DELETE:
- serverNodes.remove(path);
- LOGGER.info("watch key {} deleted, key is {}",
watchKey, path);
- continue;
- default:
- }
- }
- watcherInstanceRegisterMap.put(selectKey,
getInstanceRegisterFun.apply(serverNodes));
- }));
- final List<InstanceEntity> instanceEntities =
getInstanceRegisterFun.apply(serverNodes);
- watcherInstanceRegisterMap.put(selectKey, instanceEntities);
- return instanceEntities;
+ return client.getKeysMapByPrefix(watchKey).values().stream().map(x -> {
+ InstanceEntity instanceEntity =
GsonUtils.getInstance().fromJson(x, InstanceEntity.class);
+ instanceEntity.setUri(getURI(x, instanceEntity.getPort(),
instanceEntity.getHost()));
+ return instanceEntity;
+ }).collect(Collectors.toList());
}
@Override
diff --git
a/shenyu-registry/shenyu-registry-etcd/src/test/java/org/apache/shenyu/registry/etcd/EtcdInstanceRegisterRepositoryTest.java
b/shenyu-registry/shenyu-registry-etcd/src/test/java/org/apache/shenyu/registry/etcd/EtcdInstanceRegisterRepositoryTest.java
index b0ec00a46f..5ceb3207b1 100644
---
a/shenyu-registry/shenyu-registry-etcd/src/test/java/org/apache/shenyu/registry/etcd/EtcdInstanceRegisterRepositoryTest.java
+++
b/shenyu-registry/shenyu-registry-etcd/src/test/java/org/apache/shenyu/registry/etcd/EtcdInstanceRegisterRepositoryTest.java
@@ -17,35 +17,49 @@
package org.apache.shenyu.registry.etcd;
+import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.ClientBuilder;
+import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Watch;
import io.etcd.jetcd.lease.LeaseGrantResponse;
+import io.etcd.jetcd.watch.WatchEvent;
+import io.etcd.jetcd.watch.WatchResponse;
import org.apache.shenyu.common.exception.ShenyuException;
import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.infra.etcd.client.EtcdClient;
import org.apache.shenyu.registry.api.config.RegisterConfig;
import org.apache.shenyu.registry.api.entity.InstanceEntity;
+import org.apache.shenyu.registry.api.event.ChangedEventListener;
import org.apache.shenyu.registry.api.path.InstancePathConstants;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.mockito.MockedConstruction;
+import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import java.lang.reflect.Field;
+import java.net.URI;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
@@ -55,6 +69,8 @@ public final class EtcdInstanceRegisterRepositoryTest {
private EtcdInstanceRegisterRepository repository;
+ private EtcdClient etcdClient;
+
private final Map<String, String> etcdBroker = new HashMap<>();
/**
@@ -77,7 +93,8 @@ public final class EtcdInstanceRegisterRepositoryTest {
}
private EtcdClient mockEtcdClient() {
- EtcdClient etcdClient = mock(EtcdClient.class);
+ etcdClient = mock(EtcdClient.class);
+
when(etcdClient.getKeysMapByPrefix(anyString())).thenReturn(Collections.emptyMap());
doAnswer(invocationOnMock -> {
String key = invocationOnMock.getArgument(0);
@@ -138,25 +155,85 @@ public final class EtcdInstanceRegisterRepositoryTest {
}
/**
- * Test select instances and watcher.
+ * Test selecting instances does not create a persistent watcher.
*/
@Test
- public void testSelectInstancesAndWatcher() {
+ public void testSelectInstancesDoesNotCreateWatcher() {
+ assertTrue(repository.selectInstances("shenyu-test").isEmpty());
+ assertTrue(repository.selectInstances("shenyu-test").isEmpty());
+
+ verify(etcdClient,
times(2)).getKeysMapByPrefix(InstancePathConstants.buildInstanceParentPath("shenyu-test"));
+ verify(etcdClient, never()).watchKeyChanges(anyString(),
any(Watch.Listener.class));
+ }
+
+ /**
+ * Test each selection reads current instances and fills their URIs.
+ */
+ @Test
+ public void testSelectInstancesReadsCurrentInstances() {
InstanceEntity data = InstanceEntity.builder()
.appName("shenyu-test")
.host("shenyu-host")
.port(9195)
.build();
+ String prefix =
InstancePathConstants.buildInstanceParentPath(data.getAppName());
+ String node = InstancePathConstants.buildRealNode(prefix,
"shenyu-host:9195");
+ Map<String, String> serverNodes = new HashMap<>();
+ when(etcdClient.getKeysMapByPrefix(prefix)).thenAnswer(invocation ->
new HashMap<>(serverNodes));
- try (MockedConstruction<EtcdClient> construction =
mockConstruction(EtcdClient.class, (mock, context) -> {
- })) {
- final EtcdInstanceRegisterRepository repository = new
EtcdInstanceRegisterRepository();
- RegisterConfig config = new RegisterConfig();
- config.setServerLists("http://localhost:2379";);
- repository.init(config);
- repository.persistInstance(data);
-
repository.selectInstances(InstancePathConstants.buildInstanceParentPath());
- repository.close();
- }
+ assertTrue(repository.selectInstances(data.getAppName()).isEmpty());
+ serverNodes.put(node, GsonUtils.getInstance().toJson(data));
+ List<InstanceEntity> instances =
repository.selectInstances(data.getAppName());
+ assertEquals(1, instances.size());
+ assertEquals(data.getAppName(), instances.get(0).getAppName());
+ assertEquals(data.getHost(), instances.get(0).getHost());
+ assertEquals(data.getPort(), instances.get(0).getPort());
+ assertEquals(URI.create("http://shenyu-host:9195";),
instances.get(0).getUri());
+
+ data.setWeight(10);
+ data.setUri(URI.create("https://shenyu-host:9195";));
+ serverNodes.put(node, GsonUtils.getInstance().toJson(data));
+ instances = repository.selectInstances(data.getAppName());
+ assertEquals(1, instances.size());
+ assertEquals(10, instances.get(0).getWeight());
+ assertEquals(URI.create("https://shenyu-host:9195";),
instances.get(0).getUri());
+
+ InstanceEntity another = new InstanceEntity(data.getAppName(),
"another-host", 9196);
+ serverNodes.put(InstancePathConstants.buildRealNode(prefix,
"another-host:9196"), GsonUtils.getInstance().toJson(another));
+ assertEquals(2, repository.selectInstances(data.getAppName()).size());
+ serverNodes.remove(node);
+ instances = repository.selectInstances(data.getAppName());
+ assertEquals(1, instances.size());
+ assertEquals(URI.create("http://another-host:9196";),
instances.get(0).getUri());
+ serverNodes.clear();
+ assertTrue(repository.selectInstances(data.getAppName()).isEmpty());
+ verify(etcdClient, times(6)).getKeysMapByPrefix(prefix);
+ }
+
+ /**
+ * Test selecting instances preserves an explicit watcher and its
notifications.
+ */
+ @Test
+ public void testSelectInstancesPreservesExplicitWatcher() {
+ String key =
InstancePathConstants.buildInstanceParentPath("shenyu-test");
+ ChangedEventListener listener = mock(ChangedEventListener.class);
+ Watch.Watcher watcher = mock(Watch.Watcher.class);
+ when(etcdClient.watchKeyChanges(eq(key),
any(Watch.Listener.class))).thenReturn(watcher);
+
+ repository.watchInstances(key, listener);
+ repository.selectInstances("shenyu-test");
+ ArgumentCaptor<Watch.Listener> captor =
ArgumentCaptor.forClass(Watch.Listener.class);
+ verify(etcdClient).watchKeyChanges(eq(key), captor.capture());
+ verify(watcher, never()).close();
+ String node = key + "/shenyu-host:9195";
+ String value = "instance-data";
+ KeyValue keyValue = mock(KeyValue.class);
+ when(keyValue.getKey()).thenReturn(ByteSequence.from(node, UTF_8));
+ when(keyValue.getValue()).thenReturn(ByteSequence.from(value, UTF_8));
+ WatchResponse response = mock(WatchResponse.class);
+ when(response.getEvents()).thenReturn(Collections.singletonList(new
WatchEvent(keyValue, keyValue, WatchEvent.EventType.PUT)));
+
+ captor.getValue().onNext(response);
+ verify(listener).onEvent(node, value,
ChangedEventListener.Event.ADDED);
}
}
