Fix Dubbo ReferenceCount OOM (#12031)

Co-authored-by: earthchen <earthchen1996@gmail.com>

dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ClientsProvider.java

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.protocol.dubbo;
+
+import org.apache.dubbo.remoting.exchange.ExchangeClient;
+
+import java.util.List;
+
+public interface ClientsProvider {
+    List<? extends ExchangeClient> getClients();
+
+    void close(int timeout);
+}

dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java

@@ -37,6 +37,7 @@ import org.apache.dubbo.rpc.RpcInvocation;
 import org.apache.dubbo.rpc.protocol.AbstractInvoker;
 import org.apache.dubbo.rpc.support.RpcUtils;
 
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
@@ -46,10 +47,9 @@ import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT;
 import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
 import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
 import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.PAYLOAD;
 import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
 import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
-import static org.apache.dubbo.common.constants.CommonConstants.PAYLOAD;
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_ERROR_CLOSE_CLIENT;
 import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
 
 /**
@@ -57,7 +57,7 @@ import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
  */
 public class DubboInvoker<T> extends AbstractInvoker<T> {
 
-    private final ExchangeClient[] clients;
+    private final ClientsProvider clientsProvider;
 
     private final AtomicPositiveInteger index = new AtomicPositiveInteger();
 
@@ -68,13 +68,13 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
 
     private final int serverShutdownTimeout;
 
-    public DubboInvoker(Class<T> serviceType, URL url, ExchangeClient[] clients) {
+    public DubboInvoker(Class<T> serviceType, URL url, ClientsProvider clientsProvider) {
-        this(serviceType, url, clients, null);
+        this(serviceType, url, clientsProvider, null);
     }
 
-    public DubboInvoker(Class<T> serviceType, URL url, ExchangeClient[] clients, Set<Invoker<?>> invokers) {
+    public DubboInvoker(Class<T> serviceType, URL url, ClientsProvider clientsProvider, Set<Invoker<?>> invokers) {
         super(serviceType, url, new String[]{INTERFACE_KEY, GROUP_KEY, TOKEN_KEY});
-        this.clients = clients;
+        this.clientsProvider = clientsProvider;
         this.invokers = invokers;
         this.serverShutdownTimeout = ConfigurationUtils.getServerShutdownTimeout(getUrl().getScopeModel());
     }
@@ -87,10 +87,11 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
         inv.setAttachment(VERSION_KEY, version);
 
         ExchangeClient currentClient;
-        if (clients.length == 1) {
+        List<? extends ExchangeClient> exchangeClients = clientsProvider.getClients();
-            currentClient = clients[0];
+        if (exchangeClients.size() == 1) {
+            currentClient = exchangeClients.get(0);
         } else {
-            currentClient = clients[index.getAndIncrement() % clients.length];
+            currentClient = exchangeClients.get(index.getAndIncrement() % exchangeClients.size());
         }
         try {
             boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
@@ -143,7 +144,7 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
         if (!super.isAvailable()) {
             return false;
         }
-        for (ExchangeClient client : clients) {
+        for (ExchangeClient client : clientsProvider.getClients()) {
             if (client.isConnected() && !client.hasAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY)) {
                 //cannot write == not Available ?
                 return true;
@@ -168,14 +169,7 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
                 if (invokers != null) {
                     invokers.remove(this);
                 }
-                for (ExchangeClient client : clients) {
+                clientsProvider.close(serverShutdownTimeout);
-                    try {
-                        client.close(serverShutdownTimeout);
-                    } catch (Throwable t) {
-                        logger.warn(PROTOCOL_ERROR_CLOSE_CLIENT, "", "", t.getMessage(), t);
-                    }
-                }
-
             } finally {
                 destroyLock.unlock();
             }

dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java

@@ -19,8 +19,8 @@ package org.apache.dubbo.rpc.protocol.dubbo;
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.URLBuilder;
 import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
 import org.apache.dubbo.common.url.component.ServiceConfigURL;
-import org.apache.dubbo.common.utils.CollectionUtils;
 import org.apache.dubbo.common.utils.NetUtils;
 import org.apache.dubbo.common.utils.StringUtils;
 import org.apache.dubbo.remoting.Channel;
@@ -49,16 +49,16 @@ import org.apache.dubbo.rpc.protocol.AbstractProtocol;
 
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
 import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
@@ -66,7 +66,6 @@ import static org.apache.dubbo.common.constants.CommonConstants.LAZY_CONNECT_KEY
 import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
 import static org.apache.dubbo.common.constants.CommonConstants.STUB_EVENT_KEY;
 import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_ERROR_CLOSE_CLIENT;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_ERROR_CLOSE_SERVER;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_REFER_INVOKER;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_UNSUPPORTED;
@@ -105,9 +104,7 @@ public class DubboProtocol extends AbstractProtocol {
      * <host:port,Exchanger>
      * Map<String, List<ReferenceCountExchangeClient>
      */
-    private final Map<String, Object> referenceClientMap = new ConcurrentHashMap<>();
+    private final Map<String, SharedClientsProvider> referenceClientMap = new ConcurrentHashMap<>();
-
-    private static final Object PENDING_OBJECT = new Object();
 
     private final AtomicBoolean destroyed = new AtomicBoolean();
 
@@ -237,6 +234,7 @@ public class DubboProtocol extends AbstractProtocol {
                 return invocation;
             }
         };
+        this.frameworkModel = frameworkModel;
     }
 
     /**
@@ -415,7 +413,7 @@ public class DubboProtocol extends AbstractProtocol {
         return invoker;
     }
 
-    private ExchangeClient[] getClients(URL url) {
+    private ClientsProvider getClients(URL url) {
         int connections = url.getParameter(CONNECTIONS_KEY, 0);
         // whether to share connection
         // if not configured, connection is shared, otherwise, one connection for one service
@@ -428,17 +426,13 @@ public class DubboProtocol extends AbstractProtocol {
                 : url.getParameter(SHARE_CONNECTIONS_KEY, (String) null);
             connections = Integer.parseInt(shareConnectionsStr);
 
-            List<ReferenceCountExchangeClient> shareClients = getSharedClient(url, connections);
+            return getSharedClient(url, connections);
-            ExchangeClient[] clients = new ExchangeClient[connections];
-            Arrays.setAll(clients, shareClients::get);
-            return clients;
         }
 
-        ExchangeClient[] clients = new ExchangeClient[connections];
+        List<ExchangeClient> clients = IntStream.range(0, connections)
-        for (int i = 0; i < clients.length; i++) {
+            .mapToObj((i) -> initClient(url))
-            clients[i] = initClient(url);
+            .collect(Collectors.toList());
-        }
+        return new ExclusiveClientsProvider(clients);
-        return clients;
     }
 
     /**
@@ -448,106 +442,24 @@ public class DubboProtocol extends AbstractProtocol {
      * @param connectNum connectNum must be greater than or equal to 1
      */
     @SuppressWarnings("unchecked")
-    private List<ReferenceCountExchangeClient> getSharedClient(URL url, int connectNum) {
+    private SharedClientsProvider getSharedClient(URL url, int connectNum) {
         String key = url.getAddress();
 
-        Object clients = referenceClientMap.get(key);
+        // connectNum must be greater than or equal to 1
-        if (clients instanceof List) {
+        int expectedConnectNum = Math.max(connectNum, 1);
-            List<ReferenceCountExchangeClient> typedClients = (List<ReferenceCountExchangeClient>) clients;
+        return referenceClientMap.compute(key, (originKey, originValue) -> {
-            if (checkClientCanUse(typedClients)) {
+            if (originValue != null && originValue.increaseCount()) {
-                batchClientRefIncr(typedClients);
+                return originValue;
-                return typedClients;
-            }
-        }
-
-        List<ReferenceCountExchangeClient> typedClients = null;
-
-        synchronized (referenceClientMap) {
-            for (; ; ) {
-                // guarantee just one thread in loading condition. And Other is waiting It had finished.
-                clients = referenceClientMap.get(key);
-
-                if (clients instanceof List) {
-                    typedClients = (List<ReferenceCountExchangeClient>) clients;
-                    if (checkClientCanUse(typedClients)) {
-                        batchClientRefIncr(typedClients);
-                        return typedClients;
-                    } else {
-                        referenceClientMap.put(key, PENDING_OBJECT);
-                        break;
-                    }
-                } else if (clients == PENDING_OBJECT) {
-                    try {
-                        referenceClientMap.wait();
-                    } catch (InterruptedException ignored) {
-                    }
-                } else {
-                    referenceClientMap.put(key, PENDING_OBJECT);
-                    break;
-                }
-            }
-        }
-
-        try {
-            // connectNum must be greater than or equal to 1
-            connectNum = Math.max(connectNum, 1);
-
-            // If the clients is empty, then the first initialization is
-            if (CollectionUtils.isEmpty(typedClients)) {
-                typedClients = buildReferenceCountExchangeClientList(url, connectNum);
             } else {
-                for (int i = 0; i < typedClients.size(); i++) {
+                return new SharedClientsProvider(this, originKey, buildReferenceCountExchangeClientList(url, expectedConnectNum));
-                    ReferenceCountExchangeClient referenceCountExchangeClient = typedClients.get(i);
-                    // If there is a client in the list that is no longer available, create a new one to replace him.
-                    if (referenceCountExchangeClient == null || referenceCountExchangeClient.isClosed()) {
-                        typedClients.set(i, buildReferenceCountExchangeClient(url));
-                        continue;
-                    }
-                    referenceCountExchangeClient.incrementAndGetCount();
-                }
             }
-        } finally {
+        });
-            synchronized (referenceClientMap) {
-                if (typedClients == null) {
-                    referenceClientMap.remove(key);
-                } else {
-                    referenceClientMap.put(key, typedClients);
-                }
-                referenceClientMap.notifyAll();
-            }
-        }
-        return typedClients;
     }
 
-    /**
+    protected void scheduleRemoveSharedClient(String key, SharedClientsProvider sharedClient) {
-     * Check if the client list is all available
+        this.frameworkModel.getBeanFactory().getBean(FrameworkExecutorRepository.class)
-     *
+            .getSharedExecutor()
-     * @param referenceCountExchangeClients
+            .submit(() -> referenceClientMap.remove(key, sharedClient));
-     * @return true-available，false-unavailable
-     */
-    private boolean checkClientCanUse(List<ReferenceCountExchangeClient> referenceCountExchangeClients) {
-        if (CollectionUtils.isEmpty(referenceCountExchangeClients)) {
-            return false;
-        }
-
-        // As long as one client is not available, you need to replace the unavailable client with the available one.
-        return referenceCountExchangeClients.stream()
-            .noneMatch(referenceCountExchangeClient -> referenceCountExchangeClient == null
-                || referenceCountExchangeClient.getCount() <= 0 || referenceCountExchangeClient.isClosed());
-    }
-
-    /**
-     * Increase the reference Count if we create new invoker shares same connection, the connection will be closed without any reference.
-     *
-     * @param referenceCountExchangeClients
-     */
-    private void batchClientRefIncr(List<ReferenceCountExchangeClient> referenceCountExchangeClients) {
-        if (CollectionUtils.isEmpty(referenceCountExchangeClients)) {
-            return;
-        }
-        referenceCountExchangeClients.stream()
-            .filter(Objects::nonNull)
-            .forEach(ReferenceCountExchangeClient::incrementAndGetCount);
     }
 
     /**
@@ -648,53 +560,16 @@ public class DubboProtocol extends AbstractProtocol {
         serverMap.clear();
 
         for (String key : new ArrayList<>(referenceClientMap.keySet())) {
-            Object clients = referenceClientMap.remove(key);
+            SharedClientsProvider clients = referenceClientMap.remove(key);
-            if (clients instanceof List) {
+            clients.forceClose();
-                List<ReferenceCountExchangeClient> typedClients = (List<ReferenceCountExchangeClient>) clients;
-
-                if (CollectionUtils.isEmpty(typedClients)) {
-                    continue;
-                }
-
-                for (ReferenceCountExchangeClient client : typedClients) {
-                    closeReferenceCountExchangeClient(client);
-                }
-            }
         }
+
         PortUnificationExchanger.close();
         referenceClientMap.clear();
 
         super.destroy();
     }
 
-    /**
-     * close ReferenceCountExchangeClient
-     *
-     * @param client
-     */
-    private void closeReferenceCountExchangeClient(ReferenceCountExchangeClient client) {
-        if (client == null) {
-            return;
-        }
-
-        try {
-            if (logger.isInfoEnabled()) {
-                logger.info("Close dubbo connect: " + client.getLocalAddress() + "-->" + client.getRemoteAddress());
-            }
-
-            client.close(client.getShutdownWaitTime());
-
-            // TODO
-            /*
-             * At this time, ReferenceCountExchangeClient#client has been replaced with LazyConnectExchangeClient.
-             * Do you need to call client.close again to ensure that LazyConnectExchangeClient is also closed?
-             */
-
-        } catch (Throwable t) {
-            logger.warn(PROTOCOL_ERROR_CLOSE_CLIENT, "", "", t.getMessage(), t);
-        }
-    }
-
     /**
      * only log body in debugger mode for size & security consideration.
      *

dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ExclusiveClientsProvider.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.protocol.dubbo;
+
+import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.remoting.exchange.ExchangeClient;
+
+import java.util.List;
+
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_ERROR_CLOSE_CLIENT;
+
+public class ExclusiveClientsProvider implements ClientsProvider {
+    private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(ExclusiveClientsProvider.class);
+    private final List<ExchangeClient> clients;
+
+    public ExclusiveClientsProvider(List<ExchangeClient> clients) {
+        this.clients = clients;
+    }
+
+    @Override
+    public List<ExchangeClient> getClients() {
+        return clients;
+    }
+
+    @Override
+    public void close(int timeout) {
+        for (ExchangeClient client : clients) {
+            try {
+                client.close(timeout);
+            } catch (Throwable t) {
+                logger.warn(PROTOCOL_ERROR_CLOSE_CLIENT, "", "", t.getMessage(), t);
+            }
+        }
+    }
+}

dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/SharedClientsProvider.java

@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.protocol.dubbo;
+
+import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.CollectionUtils;
+
+import java.util.List;
+import java.util.Objects;
+
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_ERROR_CLOSE_CLIENT;
+
+public class SharedClientsProvider implements ClientsProvider {
+    private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(SharedClientsProvider.class);
+    private final DubboProtocol dubboProtocol;
+    private final String addressKey;
+    private final List<ReferenceCountExchangeClient> clients;
+
+    public SharedClientsProvider(DubboProtocol dubboProtocol, String addressKey, List<ReferenceCountExchangeClient> clients) {
+        this.dubboProtocol = dubboProtocol;
+        this.addressKey = addressKey;
+        this.clients = clients;
+    }
+
+    @Override
+    public List<ReferenceCountExchangeClient> getClients() {
+        return clients;
+    }
+
+    public synchronized boolean increaseCount() {
+        if (checkClientCanUse(clients)) {
+            batchClientRefIncr(clients);
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public synchronized void close(int timeout) {
+        for (ReferenceCountExchangeClient client : clients) {
+            try {
+                client.close(timeout);
+            } catch (Throwable t) {
+                logger.warn(PROTOCOL_ERROR_CLOSE_CLIENT, "", "", t.getMessage(), t);
+            }
+        }
+        if (!checkClientCanUse(clients)) {
+            dubboProtocol.scheduleRemoveSharedClient(addressKey, this);
+        }
+    }
+
+    public synchronized void forceClose() {
+        for (ReferenceCountExchangeClient client : clients) {
+            closeReferenceCountExchangeClient(client);
+        }
+    }
+
+    /**
+     * Check if the client list is all available
+     *
+     * @param referenceCountExchangeClients
+     * @return true-available，false-unavailable
+     */
+    private boolean checkClientCanUse(List<ReferenceCountExchangeClient> referenceCountExchangeClients) {
+        if (CollectionUtils.isEmpty(referenceCountExchangeClients)) {
+            return false;
+        }
+
+        // As long as one client is not available, you need to replace the unavailable client with the available one.
+        return referenceCountExchangeClients.stream()
+            .noneMatch(referenceCountExchangeClient -> referenceCountExchangeClient == null
+                || referenceCountExchangeClient.getCount() <= 0 || referenceCountExchangeClient.isClosed());
+    }
+
+    /**
+     * Increase the reference Count if we create new invoker shares same connection, the connection will be closed without any reference.
+     *
+     * @param referenceCountExchangeClients
+     */
+    private void batchClientRefIncr(List<ReferenceCountExchangeClient> referenceCountExchangeClients) {
+        if (CollectionUtils.isEmpty(referenceCountExchangeClients)) {
+            return;
+        }
+        referenceCountExchangeClients.stream()
+            .filter(Objects::nonNull)
+            .forEach(ReferenceCountExchangeClient::incrementAndGetCount);
+    }
+
+    /**
+     * close ReferenceCountExchangeClient
+     *
+     * @param client
+     */
+    private void closeReferenceCountExchangeClient(ReferenceCountExchangeClient client) {
+        if (client == null) {
+            return;
+        }
+
+        try {
+            if (logger.isInfoEnabled()) {
+                logger.info("Close dubbo connect: " + client.getLocalAddress() + "-->" + client.getRemoteAddress());
+            }
+
+            client.close(client.getShutdownWaitTime());
+
+            // TODO
+            /*
+             * At this time, ReferenceCountExchangeClient#client has been replaced with LazyConnectExchangeClient.
+             * Do you need to call client.close again to ensure that LazyConnectExchangeClient is also closed?
+             */
+
+        } catch (Throwable t) {
+            logger.warn(PROTOCOL_ERROR_CLOSE_CLIENT, "", "", t.getMessage(), t);
+        }
+    }
+
+}

dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvokerAvailableTest.java

@@ -36,6 +36,7 @@ import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import java.lang.reflect.Field;
+import java.util.List;
 
 import static org.apache.dubbo.common.constants.CommonConstants.SHUTDOWN_WAIT_KEY;
 
@@ -172,11 +173,12 @@ class DubboInvokerAvailableTest {
     }
 
     private ExchangeClient[] getClients(DubboInvoker<?> invoker) throws Exception {
-        Field field = DubboInvoker.class.getDeclaredField("clients");
+        Field field = DubboInvoker.class.getDeclaredField("clientsProvider");
         field.setAccessible(true);
-        ExchangeClient[] clients = (ExchangeClient[]) field.get(invoker);
+        ClientsProvider clientsProvider = (ClientsProvider) field.get(invoker);
-        Assertions.assertEquals(1, clients.length);
+        List<? extends ExchangeClient> clients = clientsProvider.getClients();
-        return clients;
+        Assertions.assertEquals(1, clients.size());
+        return clients.toArray(new ExchangeClient[0]);
     }
 
     public class DemoServiceImpl implements IDemoService {

dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClientTest.java

@@ -205,7 +205,7 @@ class ReferenceCountExchangeClientTest {
         // invoker status is available because the default value of associated lazy client's initial state is true.
         Assertions.assertTrue(helloServiceInvoker.isAvailable(), "invoker status unavailable");
 
-        // revive: initial the lazy client's exchange client again.  
+        // revive: initial the lazy client's exchange client again.
         Assertions.assertEquals("hello", helloService.hello());
 
         destroy();
@@ -285,11 +285,12 @@ class ReferenceCountExchangeClientTest {
     private List<ExchangeClient> getInvokerClientList(Invoker<?> invoker) {
         @SuppressWarnings("rawtypes") DubboInvoker dInvoker = (DubboInvoker) invoker;
         try {
-            Field clientField = DubboInvoker.class.getDeclaredField("clients");
+            Field clientField = DubboInvoker.class.getDeclaredField("clientsProvider");
             clientField.setAccessible(true);
-            ExchangeClient[] clients = (ExchangeClient[]) clientField.get(dInvoker);
+            ClientsProvider clientsProvider = (ClientsProvider) clientField.get(dInvoker);
+            List<? extends ExchangeClient> clients = clientsProvider.getClients();
 
-            List<ExchangeClient> clientList = new ArrayList<ExchangeClient>(clients.length);
+            List<ExchangeClient> clientList = new ArrayList<ExchangeClient>(clients.size());
             for (ExchangeClient client : clients) {
                 clientList.add(client);
             }