AcceptMediocrity commented on issue #16375: URL: https://github.com/apache/dubbo/issues/16375#issuecomment-5407449068
> [@AcceptMediocrity](https://github.com/AcceptMediocrity) hello bro. So far, the cause of the problem has been identified as follows: > > After a Pod in K8s is restarted, the IP address of the service instance changes. However, in some cases, the Dubbo consumer still retains the old Pod IP address, causing attempts to connect to an no longer existent address to continue. The specific trigger point lies in the service discovery notification mechanism: when Nacos/K8s sends a service discovery notification, ServiceInstancesChangedListener receives the list of new instances. The listener then retrieves the metadata based on the revision of each instance. If all versions of the metadata are failed to be retrieved during a notification, the old logic will directly return. This premature return prevents further notifications from being sent to downstream components. Although the downstream ServiceDiscoveryRegistryDirectory has the capability to clean up unused invokers using destroyUnusedInvokers(), it doesn’t receive any notifications indicating that the address has changed or been replaced. Therefore, the old IP addr ess remains in use. > > **The solution** is to remove the premature return of ServiceInstancesChangedListener when “all metadata is empty/failed”. This means that errors are still recorded, retries are still attempted, but the process continues with the reconstruction of service URLs and notification of changes in the address. In this way, even if the metadata retrieval fails temporarily, the status of the currently resolvable address is still notified. For the old service addresses, subsequent directory refreshes will return either an empty or new address, thereby triggering the elimination of the old invoker, and preventing continued connection to the old Pod IP address. Dubbo：3.2.13 org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener#doOnEvent private synchronized void doOnEvent(ServiceInstancesChangedEvent event) { if (destroyed.get() || !accept(event) || isRetryAndExpired(event)) { return; } refreshInstance(event); if (logger.isDebugEnabled()) { logger.debug(event.getServiceInstances().toString()); } Map<String, List<ServiceInstance>> revisionToInstances = new HashMap<>(); Map<ServiceInfo, Set<String>> localServiceToRevisions = new HashMap<>(); // grouping all instances of this app(service name) by revision for (Map.Entry<String, List<ServiceInstance>> entry : allInstances.entrySet()) { List<ServiceInstance> instances = entry.getValue(); for (ServiceInstance instance : instances) { String revision = getExportedServicesRevision(instance); if (revision == null || EMPTY_REVISION.equals(revision)) { if (logger.isDebugEnabled()) { logger.debug("Find instance without valid service metadata: " + instance.getAddress()); } continue; } List<ServiceInstance> subInstances = revisionToInstances.computeIfAbsent(revision, r -> new LinkedList<>()); subInstances.add(instance); } } // get MetadataInfo with revision for (Map.Entry<String, List<ServiceInstance>> entry : revisionToInstances.entrySet()) { String revision = entry.getKey(); List<ServiceInstance> subInstances = entry.getValue(); MetadataInfo metadata = subInstances.stream() .map(ServiceInstance::getServiceMetadata) .filter(Objects::nonNull) .filter(m -> revision.equals(m.getRevision())) .findFirst() .orElseGet(() -> serviceDiscovery.getRemoteMetadata(revision, subInstances)); parseMetadata(revision, metadata, localServiceToRevisions); // update metadata into each instance, in case new instance created. for (ServiceInstance tmpInstance : subInstances) { MetadataInfo originMetadata = tmpInstance.getServiceMetadata(); if (originMetadata == null || !Objects.equals(originMetadata.getRevision(), metadata.getRevision())) { tmpInstance.setServiceMetadata(metadata); } } } int emptyNum = hasEmptyMetadata(revisionToInstances); if (emptyNum != 0) { // retry every 10 seconds hasEmptyMetadata = true; if (retryPermission.tryAcquire()) { if (retryFuture != null && !retryFuture.isDone()) { // cancel last retryFuture because only one retryFuture will be canceled at destroy(). retryFuture.cancel(true); } try { retryFuture = scheduler.schedule( new AddressRefreshRetryTask(retryPermission, event.getServiceName()), 10_000L, TimeUnit.MILLISECONDS); } catch (Exception e) { logger.error( INTERNAL_ERROR, "unknown error in registry module", "", "Error submitting async retry task."); } logger.warn( INTERNAL_ERROR, "unknown error in registry module", "", "Address refresh try task submitted"); } // return if all metadata is empty, this notification will not take effect. if (emptyNum == revisionToInstances.size()) { // 1-17 - Address refresh failed. logger.error( REGISTRY_FAILED_REFRESH_ADDRESS, "metadata Server failure", "", "Address refresh failed because of Metadata Server failure, wait for retry or new address refresh event."); return; } } hasEmptyMetadata = false; Map<String, Map<Integer, Map<Set<String>, Object>>> protocolRevisionsToUrls = new HashMap<>(); Map<String, List<ProtocolServiceKeyWithUrls>> newServiceUrls = new HashMap<>(); for (Map.Entry<ServiceInfo, Set<String>> entry : localServiceToRevisions.entrySet()) { ServiceInfo serviceInfo = entry.getKey(); Set<String> revisions = entry.getValue(); Map<Integer, Map<Set<String>, Object>> portToRevisions = protocolRevisionsToUrls.computeIfAbsent(serviceInfo.getProtocol(), k -> new HashMap<>()); Map<Set<String>, Object> revisionsToUrls = portToRevisions.computeIfAbsent(serviceInfo.getPort(), k -> new HashMap<>()); Object urls = revisionsToUrls.computeIfAbsent( revisions, k -> getServiceUrlsCache( revisionToInstances, revisions, serviceInfo.getProtocol(), serviceInfo.getPort())); List<ProtocolServiceKeyWithUrls> list = newServiceUrls.computeIfAbsent(serviceInfo.getPath(), k -> new LinkedList<>()); list.add(new ProtocolServiceKeyWithUrls(serviceInfo.getProtocolServiceKey(), (List<URL>) urls)); } this.serviceUrls = newServiceUrls; this.notifyAddressChanged(); } 没有找到你改动的那块代码 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
