[I] tri和dubbo端口融合, consumer TripleInvoker和DubboInvoker覆盖问题 [dubbo]

[via GitHub]

iJIAJIA opened a new issue, #13401:
URL: https://github.com/apache/dubbo/issues/13401

   <!-- If you need to report a security issue please visit 
https://github.com/apache/dubbo/security/policy -->
   
   - [x] I have searched the [issues](https://github.com/apache/dubbo/issues) 
of this repository and believe that this is not a duplicate.
   
   ### Environment
   
   * Dubbo version: 3.2.3
   * Operating System version: win10
   * Java version: jdk 8
   
   ### Steps to reproduce this issue
   
   1. provider 使用端口融合
   ```
   dubbo:
     protocol:
       name: tri
       ext-protocol: dubbo
   ```
   ### Expected Behavior
   
org.apache.dubbo.registry.client.ServiceDiscoveryRegistryDirectory#toInvokers 
返回 TripleInvoker和DubboInvoker.
   
![image](https://github.com/apache/dubbo/assets/10057108/27b33def-1fa1-4389-af27-757603d7537b)
   
   ### Actual Behavior
   
   
org.apache.dubbo.registry.client.ServiceDiscoveryRegistryDirectory#toInvokers 
两个协议,返回了两个DubboInovker或者TripleInvoker.
   
![image](https://github.com/apache/dubbo/assets/10057108/1fc71198-54e7-4e26-926e-7284f820e0c8)
   
   
   <!-- What actually happens? -->
   **原因**
   ```
   private Map<ProtocolServiceKeyWithAddress, Invoker<T>> 
toInvokers(Map<ProtocolServiceKeyWithAddress, Invoker<T>> oldUrlInvokerMap, 
List<URL> urls) {
           
           for (URL url : urls) {
               // 由于provider启用了端口复用, tri和dubbo两个实例端口相同
               InstanceAddressURL instanceAddressURL = (InstanceAddressURL) url;
               
               // filter all the service available (version wildcard, group 
wildcard, protocol wildcard)
               int port = instanceAddressURL.getPort();
               List<ProtocolServiceKey> matchedProtocolServiceKeys = 
instanceAddressURL.getMetadataInfo()
                   .getMatchedServiceInfos(consumerProtocolServiceKey)
                   .stream()
                   // 这里进行实例元数据匹配时, 会匹配tri和dubbo两个实例, 仅靠端口做进一步过滤时, 
没能过滤掉不匹配的protocol协议.
                   .filter(serviceInfo -> serviceInfo.getPort() <= 0 || 
serviceInfo.getPort() == port)
                   .map(MetadataInfo.ServiceInfo::getProtocolServiceKey)
                   .collect(Collectors.toList());
   
               // 一个protocol匹配了两个ProtocolServiceKey, 在下一个instanceAddressUrl循环时, 
会覆盖前一次创建的Invoker.
               for (ProtocolServiceKey matchedProtocolServiceKey : 
matchedProtocolServiceKeys) {
                   
                   ProtocolServiceKeyWithAddress protocolServiceKeyWithAddress 
= new ProtocolServiceKeyWithAddress(matchedProtocolServiceKey, 
instanceAddressURL.getAddress());
                   Invoker<T> invoker = oldUrlInvokerMap == null ? null : 
oldUrlInvokerMap.get(protocolServiceKeyWithAddress);
                   if (invoker == null || urlChanged(invoker, 
instanceAddressURL, matchedProtocolServiceKey)) { // Not in the cache, refer 
again
                       try {
                           iboolean enabled;
                           if (instanceAddressURL.hasParameter(DISABLED_KEY)) {
                               enabled = 
!instanceAddressURL.getParameter(DISABLED_KEY, false);
                           } else {
                               enabled = 
instanceAddressURL.getParameter(ENABLED_KEY, true);
                           }
                           if (enabled) {
                               if (shouldWrap) {
                                   ....
                                   //  这里遍历List<ProtocolServiceKey>  
instanceAddressURL进行protocol创建, 
                                   invoker = new 
InstanceWrappedInvoker<>(protocol.refer(serviceType, instanceAddressURL), 
newConsumerUrl, matchedProtocolServiceKey);
                               } else {
                                   invoker = protocol.refer(serviceType, 
instanceAddressURL);
                               }
                       if (invoker != null) { // Put new invoker in cache
                           newUrlInvokerMap.put(protocolServiceKeyWithAddress, 
invoker);
                       }
                   } else {
                       newUrlInvokerMap.put(protocolServiceKeyWithAddress, 
invoker);
                       oldUrlInvokerMap.remove(protocolServiceKeyWithAddress, 
invoker);
                   }
               }
           }
           return newUrlInvokerMap;
       }
   ```
   
![image](https://github.com/apache/dubbo/assets/10057108/6cdee271-5273-44cb-b2c1-3241728831ae)
   
   
   **解决建议**
   ```
   int port = instanceAddressURL.getPort();
               String instanceProtocol = instanceAddressURL.getProtocol();
               List<ProtocolServiceKey> matchedProtocolServiceKeys = 
instanceAddressURL.getMetadataInfo()
                   .getMatchedServiceInfos(consumerProtocolServiceKey)
                   .stream()
                   .filter(serviceInfo -> serviceInfo.getPort() <= 0 || 
serviceInfo.getPort() == port)
                   // 额外添加基于instanceAddressUrl的protocol过滤条件, 
修复端口复用场景下匹配不同协议的ProtocolServiceKey问题
                   .filter(serviceInfo -> 
Objects.equals(serviceInfo.getProtocol(), instanceProtocol))
                   .map(MetadataInfo.ServiceInfo::getProtocolServiceKey)
                   .collect(Collectors.toList());
   ```
   
   


-- 
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: notifications-unsubscr...@dubbo.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscr...@dubbo.apache.org
For additional commands, e-mail: notifications-h...@dubbo.apache.org