decrements total queued mutation size when update idles (#5236)

fixes #5235
apache · Jan 8, 2025 · d5756e8 · d5756e8
1 parent 36c9740
commit d5756e8
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 2 deletions.
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
@@ -258,7 +258,21 @@ public long startUpdate(TInfo tinfo, TCredentials credentials, TDurability tdura
 
     UpdateSession us =
         new UpdateSession(new TservConstraintEnv(server.getContext(), security, credentials),
-            credentials, durability);
+            credentials, durability) {
+          @Override
+          public boolean cleanup() {
+            // This is called when a client abandons a session. When this happens need to decrement
+            // any queued mutations.
+            if (queuedMutationSize > 0) {
+              log.trace(
+                  "cleaning up abandoned update session, decrementing totalQueuedMutationSize by {}",
+                  queuedMutationSize);
+              server.updateTotalQueuedMutationSize(-queuedMutationSize);
+              queuedMutationSize = 0;
+            }
+            return true;
+          }
+        };
     return server.sessionManager.createSession(us, false);
   }
 

diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -417,7 +417,12 @@ public void run() {
   }
 
   public long updateTotalQueuedMutationSize(long additionalMutationSize) {
-    return totalQueuedMutationSize.addAndGet(additionalMutationSize);
+    var newTotal = totalQueuedMutationSize.addAndGet(additionalMutationSize);
+    if (log.isTraceEnabled()) {
+      log.trace("totalQueuedMutationSize is now {} after adding {}", newTotal,
+          additionalMutationSize);
+    }
+    return newTotal;
   }
 
   @Override