Skip to content

bugfix：修复 nacos client 不会回调 listener 方法的问题 #13341

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: develop
Choose a base branch
from
+32 −26
Open

bugfix：修复 nacos client 不会回调 listener 方法的问题 #13341

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 13 additions & 7 deletions client/src/main/java/com/alibaba/nacos/client/naming/cache/ServiceInfoHolder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ public class ServiceInfoHolder implements Closeable {
private String notifierEventScope;

private boolean enableClientMetrics = true;

public ServiceInfoHolder(String namespace, String notifierEventScope, NacosClientProperties properties) {
cacheDir = CacheDirUtil.initCacheDir(namespace, properties);
instancesDiffer = new InstancesDiffer();
Expand Down Expand Up @@ -109,10 +109,10 @@ public ServiceInfo getServiceInfo(final String serviceName, final String groupNa
* @param json service json
* @return service info
*/
public ServiceInfo processServiceInfo(String json) {
public ServiceInfo processServiceInfo(String json, boolean forceNotify) {
ServiceInfo serviceInfo = JacksonUtils.toObj(json, ServiceInfo.class);
serviceInfo.setJsonFromServer(json);
return processServiceInfo(serviceInfo);
return processServiceInfo(serviceInfo, forceNotify);
}

/**
Expand All @@ -121,7 +121,7 @@ public ServiceInfo processServiceInfo(String json) {
* @param serviceInfo new service info
* @return service info
*/
public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
public ServiceInfo processServiceInfo(ServiceInfo serviceInfo, boolean forceNotify) {
String serviceKey = serviceInfo.getKeyWithoutClusters();
if (serviceKey == null) {
NAMING_LOGGER.warn("process service info but serviceKey is null, service host: {}",
Expand All @@ -133,23 +133,29 @@ public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
//empty or error push, just ignore
NAMING_LOGGER.warn("process service info but found empty or error push, serviceKey: {}, "
+ "pushEmptyProtection: {}, hosts: {}", serviceKey, pushEmptyProtection, serviceInfo.getHosts());
if (forceNotify) {
InstancesDiff diff = getServiceInfoDiff(null, oldService);
NotifyCenter.publishEvent(
new InstancesChangeEvent(notifierEventScope, oldService.getName(), oldService.getGroupName(),
oldService.getClusters(), oldService.getHosts(), diff));
}
Comment on lines +136 to +141
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

修复ServiceInfoHolder在serviceKey为null时可能触发无效事件通知的问题

🟡 Major | 🐞 Bugs

📋 问题详情

在processServiceInfo方法中，当serviceKey为null时，新增的forceNotify分支直接使用oldService参数调用NotifyCenter.publishEvent。但此时oldService可能未初始化（如首次调用），导致传入无效参数。这可能导致事件监听器收到不完整的数据，影响下游服务发现逻辑。

💡 解决方案

需要确保oldService非空后再触发事件：

+ if (oldService != null && forceNotify) {
- if (forceNotify) {
    InstancesDiff diff = getServiceInfoDiff(null, oldService);
    NotifyCenter.publishEvent(
            new InstancesChangeEvent(notifierEventScope, oldService.getName(), oldService.getGroupName(),
                    oldService.getClusters(), oldService.getHosts(), diff));
+ }

确保在首次调用时oldService未初始化时不会触发无效事件。

您的反馈对我们很重要！(建议右键在新标签页中打开以下链接)

有用意见👍 ｜ 无用意见👎 ｜ 错误意见❌

return oldService;
}
serviceInfoMap.put(serviceKey, serviceInfo);
InstancesDiff diff = getServiceInfoDiff(oldService, serviceInfo);
if (StringUtils.isBlank(serviceInfo.getJsonFromServer())) {
serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo));
}

if (enableClientMetrics) {
try {
MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size());
} catch (Throwable t) {
NAMING_LOGGER.error("Failed to update metrics for service info map size", t);
}
}
if (diff.hasDifferent()) {

if (diff.hasDifferent() || forceNotify) {
NAMING_LOGGER.info("current ips:({}) service: {} -> {}", serviceInfo.ipCount(), serviceKey,
JacksonUtils.toJson(serviceInfo.getHosts()));

Expand Down
4 changes: 2 additions & 2 deletions client/src/main/java/com/alibaba/nacos/client/naming/core/ServiceInfoUpdateService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,7 @@ public void run() {
ServiceInfo serviceObj = serviceInfoHolder.getServiceInfoMap().get(serviceKey);
if (serviceObj == null) {
serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, false);
serviceInfoHolder.processServiceInfo(serviceObj);
serviceInfoHolder.processServiceInfo(serviceObj, false);
// TODO multiple time can be configured.
delayTime = serviceObj.getCacheMillis() * DEFAULT_UPDATE_CACHE_TIME_MULTIPLE;
lastRefTime = serviceObj.getLastRefTime();
Expand All @@ -202,7 +202,7 @@ public void run() {

if (serviceObj.getLastRefTime() <= lastRefTime) {
serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, false);
serviceInfoHolder.processServiceInfo(serviceObj);
serviceInfoHolder.processServiceInfo(serviceObj, false);
}
lastRefTime = serviceObj.getLastRefTime();
if (CollectionUtils.isEmpty(serviceObj.getHosts())) {
Expand Down
2 changes: 1 addition & 1 deletion client/src/main/java/com/alibaba/nacos/client/naming/remote/NamingClientProxyDelegate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,7 +172,7 @@ public ServiceInfo subscribe(String serviceName, String groupName, String cluste
if (null == result || !isSubscribed(serviceName, groupName, clusters)) {
result = grpcClientProxy.subscribe(serviceName, groupName, clusters);
}
serviceInfoHolder.processServiceInfo(result);
serviceInfoHolder.processServiceInfo(result, true);
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

修正processServiceInfo方法调用参数传递问题

🔴 Blocker | 🐞 Bugs

📋 问题详情

在subscribe方法中调用processServiceInfo时传递了false作为forceNotify参数。根据设计意图，订阅操作应强制触发监听器通知，此处应传递true以确保事件触发。

💡 解决方案

应使用true确保订阅时触发事件：

- serviceInfoHolder.processServiceInfo(result, false);
+ serviceInfoHolder.processServiceInfo(result, true);

确保客户端订阅时强制通知监听器更新服务信息。

您的反馈对我们很重要！(建议右键在新标签页中打开以下链接)

有用意见👍 ｜ 无用意见👎 ｜ 错误意见❌

return result;
}

Expand Down
2 changes: 1 addition & 1 deletion ...t/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ public NamingPushRequestHandler(ServiceInfoHolder serviceInfoHolder) {
public Response requestReply(Request request, Connection connection) {
if (request instanceof NotifySubscriberRequest) {
NotifySubscriberRequest notifyRequest = (NotifySubscriberRequest) request;
serviceInfoHolder.processServiceInfo(notifyRequest.getServiceInfo());
serviceInfoHolder.processServiceInfo(notifyRequest.getServiceInfo(), false);
return new NotifySubscriberResponse();
}
return null;
Expand Down
26 changes: 13 additions & 13 deletions client/src/test/java/com/alibaba/nacos/client/naming/cache/ServiceInfoHolderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ void testProcessServiceInfo() {
hosts.add(instance2);
info.setHosts(hosts);

ServiceInfo actual1 = holder.processServiceInfo(info);
ServiceInfo actual1 = holder.processServiceInfo(info, false);
assertEquals(info, actual1);

Instance newInstance1 = createInstance("1.1.1.1", 1);
Expand All @@ -97,7 +97,7 @@ void testProcessServiceInfo() {
ServiceInfo info2 = new ServiceInfo("a@@b@@c");
info2.setHosts(hosts2);

ServiceInfo actual2 = holder.processServiceInfo(info2);
ServiceInfo actual2 = holder.processServiceInfo(info2, false);
assertEquals(info2, actual2);
}

Expand All @@ -116,7 +116,7 @@ void testProcessServiceInfoEnableClientMetricsTrue() {
try (MockedStatic<MetricsMonitor> mockedMetricsMonitor = Mockito.mockStatic(MetricsMonitor.class)) {
mockedMetricsMonitor.when(MetricsMonitor::getServiceInfoMapSizeMonitor).thenReturn(mockGaugeChild);

holder.processServiceInfo(info);
holder.processServiceInfo(info, false);

verify(mockGaugeChild, times(1)).set(1);
}
Expand All @@ -134,7 +134,7 @@ void testProcessServiceInfoEnableClientMetricsFalse() {
info.setHosts(hosts);

try (MockedStatic<MetricsMonitor> mockedMetricsMonitor = Mockito.mockStatic(MetricsMonitor.class)) {
holder.processServiceInfo(info);
holder.processServiceInfo(info, false);

mockedMetricsMonitor.verify(MetricsMonitor::getServiceInfoMapSizeMonitor, never());
}
Expand All @@ -156,7 +156,7 @@ void testProcessServiceInfoEnableClientMetricsNotSet() {
try (MockedStatic<MetricsMonitor> mockedMetricsMonitor = Mockito.mockStatic(MetricsMonitor.class)) {
mockedMetricsMonitor.when(MetricsMonitor::getServiceInfoMapSizeMonitor).thenReturn(mockGaugeChild);

holder.processServiceInfo(info);
holder.processServiceInfo(info, false);

verify(mockGaugeChild, times(1)).set(1);
}
Expand All @@ -180,7 +180,7 @@ void testProcessServiceInfoSetThrowsException() {
mockedMetricsMonitor.when(MetricsMonitor::getServiceInfoMapSizeMonitor).thenReturn(mockGaugeChild);
doThrow(exception).when(mockGaugeChild).set(anyInt());

ServiceInfo actual2 = holder.processServiceInfo(info);
ServiceInfo actual2 = holder.processServiceInfo(info, false);

assertEquals(info, actual2);
}
Expand Down Expand Up @@ -208,7 +208,7 @@ private Instance createInstance(String ip, int port) {
void testProcessServiceInfo2() {
String json = "{\"groupName\":\"a\",\"name\":\"b\",\"clusters\":\"c\"}";

ServiceInfo actual = holder.processServiceInfo(json);
ServiceInfo actual = holder.processServiceInfo(json, false);
ServiceInfo expect = new ServiceInfo("a@@b@@c");
expect.setJsonFromServer(json);
assertEquals(expect.getKey(), actual.getKey());
Expand All @@ -227,19 +227,19 @@ void testProcessServiceInfoWithPushEmpty() throws NacosException {
nacosClientProperties.setProperty(PropertyKeyConst.NAMING_PUSH_EMPTY_PROTECTION, "true");
holder.shutdown();
holder = new ServiceInfoHolder("aa", "scope-001", nacosClientProperties);
holder.processServiceInfo(oldInfo);
holder.processServiceInfo(oldInfo, false);

ServiceInfo newInfo = new ServiceInfo("a@@b@@c");

final ServiceInfo actual = holder.processServiceInfo(newInfo);
final ServiceInfo actual = holder.processServiceInfo(newInfo, false);

assertEquals(oldInfo.getKey(), actual.getKey());
assertEquals(2, actual.getHosts().size());
}

@Test
void testProcessNullServiceInfo() {
assertNull(holder.processServiceInfo(new ServiceInfo()));
assertNull(holder.processServiceInfo(new ServiceInfo(), false));
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

补充测试用例覆盖forceNotify为true且无差异的场景

🟡 Major | 📝 Unit test coverage

📋 问题详情

现有测试用例未覆盖当forceNotify为true但实例差异为false时触发事件的场景，可能导致该逻辑未经过充分测试。

💡 解决方案

新增测试用例验证forceNotify强制触发：

+ @Test
+ void testForceNotifyWithoutDiff() {
+     ServiceInfo oldInfo = new ServiceInfo("test");
+     serviceInfoHolder.processServiceInfo(oldInfo, false);
+     ServiceInfo newInfo = new ServiceInfo("test");
+     serviceInfoHolder.processServiceInfo(newInfo, true);
+     verify(NotifyCenter).publishEvent(any(InstancesChangeEvent.class));
+ }

验证即使无差异时forceNotify=true仍会触发事件。

您的反馈对我们很重要！(建议右键在新标签页中打开以下链接)

有用意见👍 ｜ 无用意见👎 ｜ 错误意见❌

}

@Test
Expand All @@ -252,10 +252,10 @@ void testProcessServiceInfoForOlder() {
hosts.add(instance2);
info.setHosts(hosts);
info.setLastRefTime(System.currentTimeMillis());
holder.processServiceInfo(info);
holder.processServiceInfo(info, false);
ServiceInfo olderInfo = new ServiceInfo("a@@b@@c");
olderInfo.setLastRefTime(0L);
final ServiceInfo actual = holder.processServiceInfo(olderInfo);
final ServiceInfo actual = holder.processServiceInfo(olderInfo, false);
assertEquals(olderInfo, actual);
}

Expand All @@ -267,7 +267,7 @@ void testGetServiceInfo() {
hosts.add(instance1);
info.setHosts(hosts);

ServiceInfo expect = holder.processServiceInfo(info);
ServiceInfo expect = holder.processServiceInfo(info, false);
String serviceName = "b";
String groupName = "a";
ServiceInfo actual = holder.getServiceInfo(serviceName, groupName);
Expand Down
2 changes: 1 addition & 1 deletion ...t/src/test/java/com/alibaba/nacos/client/naming/remote/NamingClientProxyDelegateTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,7 +298,7 @@ void testSubscribe() throws NacosException {
ServiceInfo actual = delegate.subscribe(serviceName, groupName, clusters);
assertEquals(info, actual);
verify(mockGrpcClient, times(1)).subscribe(serviceName, groupName, clusters);
verify(holder, times(1)).processServiceInfo(info);
verify(holder, times(1)).processServiceInfo(info, false);

}

Expand Down
2 changes: 1 addition & 1 deletion ...c/test/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushRequestHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ void testRequestReply() {
Response response = handler.requestReply(req, new TestConnection(new RpcClient.ServerInfo()));
//then
assertTrue(response instanceof NotifySubscriberResponse);
verify(holder, times(1)).processServiceInfo(info);
verify(holder, times(1)).processServiceInfo(info, false);
}

@Test
Expand Down
Loading