If the client already disconnect, ignore this request to avoid NPE. (#…

…6175)

* if the client already disconnect, ignore this request to avoid npe.

* unify code with DistroClientDataProcessor.

* print warn info when client already disconnect.

* transfer client check logic to method.

* if client is not ephemeral, is illegal.

common/src/main/java/com/alibaba/nacos/common/remote/client/RpcClient.java

@@ -89,7 +89,7 @@ public abstract class RpcClient implements Closeable {
     protected ClientAbilities clientAbilities;
 
     /**
-     * default keep alive time 10s.
+     * default keep alive time 5s.
      */
     private long keepAliveTime = 5000L;
 
@@ -176,7 +176,6 @@ public RpcClient keepAlive(long keepAliveTime, TimeUnit timeUnit) {
         this.keepAliveTime = keepAliveTime * timeUnit.toMillis(keepAliveTime);
         LoggerUtils.printIfInfoEnabled(LOGGER, "[{}]RpcClient init keepalive time, keepAliveTimeMillis={}", name,
                 keepAliveTime);
-
         return this;
     }
 
@@ -706,7 +705,7 @@ public Response request(Request request, long timeoutMills) throws NacosExceptio
             throw (exceptionThrow instanceof NacosException) ? (NacosException) exceptionThrow
                     : new NacosException(SERVER_ERROR, exceptionThrow);
         } else {
-            throw new NacosException(SERVER_ERROR, "Request fail,Unknown Error");
+            throw new NacosException(SERVER_ERROR, "Request fail, unknown Error");
         }
     }
 

naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/impl/EphemeralClientOperationServiceImpl.java

@@ -49,6 +49,9 @@ public EphemeralClientOperationServiceImpl(ClientManagerDelegate clientManager)
     public void registerInstance(Service service, Instance instance, String clientId) {
         Service singleton = ServiceManager.getInstance().getSingleton(service);
         Client client = clientManager.getClient(clientId);
+        if (!clientIsLegal(client, clientId)) {
+            return;
+        }
         InstancePublishInfo instanceInfo = getPublishInfo(instance);
         client.addServiceInstance(singleton, instanceInfo);
         client.setLastUpdatedTime();
@@ -65,6 +68,9 @@ public void deregisterInstance(Service service, Instance instance, String client
         }
         Service singleton = ServiceManager.getInstance().getSingleton(service);
         Client client = clientManager.getClient(clientId);
+        if (!clientIsLegal(client, clientId)) {
+            return;
+        }
         InstancePublishInfo removedInstance = client.removeServiceInstance(singleton);
         client.setLastUpdatedTime();
         if (null != removedInstance) {
@@ -78,6 +84,9 @@ public void deregisterInstance(Service service, Instance instance, String client
     public void subscribeService(Service service, Subscriber subscriber, String clientId) {
         Service singleton = ServiceManager.getInstance().getSingletonIfExist(service).orElse(service);
         Client client = clientManager.getClient(clientId);
+        if (!clientIsLegal(client, clientId)) {
+            return;
+        }
         client.addServiceSubscriber(singleton, subscriber);
         client.setLastUpdatedTime();
         NotifyCenter.publishEvent(new ClientOperationEvent.ClientSubscribeServiceEvent(singleton, clientId));
@@ -87,9 +96,23 @@ public void subscribeService(Service service, Subscriber subscriber, String clie
     public void unsubscribeService(Service service, Subscriber subscriber, String clientId) {
         Service singleton = ServiceManager.getInstance().getSingletonIfExist(service).orElse(service);
         Client client = clientManager.getClient(clientId);
+        if (!clientIsLegal(client, clientId)) {
+            return;
+        }
         client.removeServiceSubscriber(singleton);
         client.setLastUpdatedTime();
         NotifyCenter.publishEvent(new ClientOperationEvent.ClientUnsubscribeServiceEvent(singleton, clientId));
-
+    }
+
+    private boolean clientIsLegal(Client client, String clientId) {
+        if (client == null) {
+            Loggers.SRV_LOG.warn("Client connection {} already disconnect", clientId);
+            return false;
+        }
+        if (!client.isEphemeral()) {
+            Loggers.SRV_LOG.warn("Client connection {} type is not ephemeral", clientId);
+            return false;
+        }
+        return true;
     }
 }