[GitHub] [accumulo] dlmarion commented on a diff in pull request #3733: ensures no writes happen after batch writer closes

["dlmarion (via GitHub)" <gi...@apache.org>]

dlmarion commented on code in PR #3733:
URL: https://github.com/apache/accumulo/pull/3733#discussion_r1307966608


##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchWriter.java:
##########
@@ -1028,14 +1022,133 @@ private MutationSet sendMutationsToTabletServer(String location,
         timeoutTracker.errorOccured();
         throw new IOException(e);
       } catch (TApplicationException tae) {
+        // no need to bother closing session in this case
+        usid = null;
         updateServerErrors(location, tae);
         throw new AccumuloServerException(location, tae);
       } catch (ThriftSecurityException e) {
+        // no need to bother closing session in this case
+        usid = null;
         updateAuthorizationFailures(
             tabMuts.keySet().stream().collect(toMap(identity(), ke -> e.code)));
         throw new AccumuloSecurityException(e.user, e.code, e);
       } catch (TException e) {
         throw new IOException(e);
+      } finally {
+        if (usid != null) {
+          // There is an open session, must close it before the batchwriter closes or writes could
+          // happen after the batch writer closes. See #3721. Queuing a task instead of executing
+          // the code here because throwing exceptions in a finally block makes the code hard to
+          // reason about. Queue is less likely to throw an exception.
+          sendThreadPool.execute(new CloseSessionTask(location, usid));
+        }
+      }
+    }
+
+    class CloseSessionTask implements Runnable {
+
+      private final String location;
+      private final long usid;
+
+      CloseSessionTask(String location, Long usid) {
+        this.location = location;
+        this.usid = usid;
+        synchronized (TabletServerBatchWriter.this) {
+          if (!failedSessions.add(new Pair<>(usid, location))) {
+            throw new IllegalStateException("Duplicate session " + location + " " + usid);
+          }
+        }
+
+      }
+
+      @Override
+      public void run() {
+        try {
+          closeSession();
+        } catch (InterruptedException | RuntimeException | ThriftSecurityException e) {
+          updateUnknownErrors("Failed to close session " + location + " " + usid, e);
+        } finally {
+          synchronized (TabletServerBatchWriter.this) {
+            if (!failedSessions.remove(new Pair<>(usid, location))) {
+              throw new IllegalStateException("Session missing " + location + " " + usid);
+            }
+            TabletServerBatchWriter.this.notifyAll();
+          }
+        }
+      }
+
+      /**
+       * Checks if there is a lock held by a tserver at a specific host and port.
+       */
+      private boolean isALockHeld(String tserver) {
+        var root = context.getZooKeeperRoot() + Constants.ZTSERVERS;
+        var zLockPath = ServiceLock.path(root + "/" + tserver);
+        return ServiceLock.getSessionId(context.getZooCache(), zLockPath) != 0;
+      }
+
+      private void closeSession() throws InterruptedException, ThriftSecurityException {
+
+        Retry retry = Retry.builder().infiniteRetries().retryAfter(100, MILLISECONDS)

Review Comment:
   Not sure if it will help, but I created RetryableThriftCall as part of the external compaction code. Example [here](https://github.com/apache/accumulo/blob/main/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java#L359)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org