Remove redundant clientSession

apache · Dec 2, 2022 · 8aa0ee1 · 8aa0ee1
1 parent 5d9e8c5
commit 8aa0ee1
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 0 deletions.
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
@@ -359,6 +359,15 @@ private void releaseClientSessions() {
         }
     }
 
+    public void removeClientSession(Endpoints endpoints, ClientSessionImpl clientSession) {
+        sessionsLock.writeLock().lock();
+        try {
+            sessionsTable.remove(endpoints, clientSession);
+        } finally {
+            sessionsLock.writeLock().unlock();
+        }
+    }
+
     private ClientSessionImpl getClientSession(Endpoints endpoints) throws ClientException {
         sessionsLock.readLock().lock();
         try {

diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientSessionImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientSessionImpl.java
@@ -65,6 +65,7 @@ private void renewRequestObserver() {
             if (sessionHandler.isEndpointsDeprecated(endpoints)) {
                 log.info("Endpoints is deprecated, no longer to renew requestObserver, endpoints={}, clientId={}",
                     endpoints, clientId);
+                sessionHandler.removeClientSession(endpoints, this);
                 return;
             }
             log.info("Try to renew requestObserver, endpoints={}, clientId={}", endpoints, clientId);

diff --git a/...ent/src/main/java/org/apache/rocketmq/client/java/impl/producer/ClientSessionHandler.java b/...ent/src/main/java/org/apache/rocketmq/client/java/impl/producer/ClientSessionHandler.java
@@ -25,6 +25,7 @@
 import io.grpc.stub.StreamObserver;
 import java.util.concurrent.ScheduledExecutorService;
 import org.apache.rocketmq.client.apis.ClientException;
+import org.apache.rocketmq.client.java.impl.ClientSessionImpl;
 import org.apache.rocketmq.client.java.misc.ClientId;
 import org.apache.rocketmq.client.java.route.Endpoints;
 
@@ -47,6 +48,11 @@ public interface ClientSessionHandler {
      */
     boolean isEndpointsDeprecated(Endpoints endpoints);
 
+    /**
+     * Remove client session.
+     */
+    void removeClientSession(Endpoints endpoints, ClientSessionImpl clientSession);
+
     /**
      * Indicates the client identifier.
      *