fix

apache · Oct 23, 2024 · 2f983ea · 2f983ea
1 parent a5e02ae
commit 2f983ea
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 4 deletions.
diff --git a/java-client/src/main/java/org/apache/pegasus/rpc/async/ReplicaSession.java b/java-client/src/main/java/org/apache/pegasus/rpc/async/ReplicaSession.java
@@ -299,10 +299,16 @@ void markSessionDisconnect() {
 
   // After the authentication is reset, a new Negotiation would be launched.
   private void resetAuth() {
+    int pendingSize;
     synchronized (authPendingSend) {
       authSucceed = false;
-      authPendingSend.clear();
+      pendingSize = authPendingSend.size();
     }
+
+    logger.info(
+        "authentication is reset for session {}, with still {} request entries pending",
+        name(),
+        pendingSize);
   }
 
   // Notify the RPC sender if failure occurred.
@@ -405,6 +411,10 @@ public void onAuthSucceed() {
       authPendingSend.clear();
     }
 
+    logger.info(
+        "authentication is successful for session {}, then {} pending request entries would be sent",
+        name(),
+        authPendingSend.size());
     sendPendingRequests(swappedPendingSend, fields);
   }
 

diff --git a/java-client/src/test/java/org/apache/pegasus/rpc/async/ReplicaSessionTest.java b/java-client/src/test/java/org/apache/pegasus/rpc/async/ReplicaSessionTest.java
@@ -60,9 +60,12 @@ public class ReplicaSessionTest {
 
   @BeforeEach
   public void before(TestInfo testInfo) throws Exception {
-    final String metaList = "127.0.0.1:34601,127.0.0.1:34602,127.0.0.1:34603";
-    manager = new ClusterManager(ClientOptions.builder().metaServers(metaList).build());
-    System.out.println("test started: " + testInfo.getDisplayName());
+    manager =
+        new ClusterManager(
+            ClientOptions.builder()
+                .metaServers("127.0.0.1:34601,127.0.0.1:34602,127.0.0.1:34603")
+                .build());
+    logger.info("test started: {}", testInfo.getDisplayName());
   }
 
   @AfterEach
@@ -323,6 +326,11 @@ public void testSessionAuth() throws InterruptedException, ExecutionException {
     queryCfgEntry.callback = cb;
     queryCfgEntry.timeoutTask = null;
 
+    // Also insert into pendingResponse since ReplicaSession#sendPendingRequests() would check
+    // sequenceId in it.
+    assertTrue(rs.pendingResponse.isEmpty());
+    rs.pendingResponse.put(queryCfgEntry.sequenceId, queryCfgEntry);
+
     // Initially session has not been authenticated.
     assertTrue(rs.tryPendRequest(queryCfgEntry));