From 3ed7cd2ebe55463ae6a7f08bb0bdf60dab775692 Mon Sep 17 00:00:00 2001
From: David Turner <david.turner@elastic.co>
Date: Mon, 3 Oct 2022 11:32:06 +0100
Subject: [PATCH 1/2] Ensure cleanups succeed in JoinValidationService

Today we sort of assume that cleanups succeed in the
`JoinValidationService`. A failure in these places might explain the
leaks seen in #90576 and #89712. It's not obvious that anything can fail
here but let's make sure.
---
 .../coordination/JoinValidationService.java   | 46 ++++++++++++-------
 1 file changed, 30 insertions(+), 16 deletions(-)

diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/JoinValidationService.java b/server/src/main/java/org/elasticsearch/cluster/coordination/JoinValidationService.java
index 95411827737eb..3c101c74f4381 100644
--- a/server/src/main/java/org/elasticsearch/cluster/coordination/JoinValidationService.java
+++ b/server/src/main/java/org/elasticsearch/cluster/coordination/JoinValidationService.java
@@ -229,27 +229,41 @@ private void processNextItem() {
         try {
             nextItem.run();
         } finally {
-            final var remaining = queueSize.decrementAndGet();
-            assert remaining >= 0;
-            if (remaining > 0) {
-                runProcessor();
+            var cleanupSuccess = false;
+            try {
+                final var remaining = queueSize.decrementAndGet();
+                assert remaining >= 0;
+                if (remaining > 0) {
+                    runProcessor();
+                }
+                cleanupSuccess = true;
+            } finally {
+                assert cleanupSuccess;
             }
         }
     }
 
     private void onShutdown() {
-        // shutting down when enqueueing the next processor run which means there is no active processor so it's safe to clear out the
-        // cache ...
-        cacheClearer.run();
-
-        // ... and drain the queue
-        do {
-            final var nextItem = queue.poll();
-            assert nextItem != null;
-            if (nextItem != cacheClearer) {
-                nextItem.onFailure(new NodeClosedException(transportService.getLocalNode()));
-            }
-        } while (queueSize.decrementAndGet() > 0);
+        var success = false;
+
+        try {
+            // shutting down when enqueueing the next processor run which means there is no active processor so it's safe to clear out the
+            // cache ...
+            cacheClearer.run();
+
+            // ... and drain the queue
+            do {
+                final var nextItem = queue.poll();
+                assert nextItem != null;
+                if (nextItem != cacheClearer) {
+                    nextItem.onFailure(new NodeClosedException(transportService.getLocalNode()));
+                }
+            } while (queueSize.decrementAndGet() > 0);
+
+            success = true;
+        } finally {
+            assert success;
+        }
     }
 
     private final AbstractRunnable cacheClearer = new AbstractRunnable() {

From 3f6c2e16eca8d973b9379132344899a1f6a4d81d Mon Sep 17 00:00:00 2001
From: David Turner <david.turner@elastic.co>
Date: Wed, 2 Nov 2022 14:10:14 +0000
Subject: [PATCH 2/2] can just catch Exception here

---
 .../coordination/JoinValidationService.java    | 18 ++++++++----------
 1 file changed, 8 insertions(+), 10 deletions(-)

diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/JoinValidationService.java b/server/src/main/java/org/elasticsearch/cluster/coordination/JoinValidationService.java
index 4028b976c5252..dee938d5e0bb6 100644
--- a/server/src/main/java/org/elasticsearch/cluster/coordination/JoinValidationService.java
+++ b/server/src/main/java/org/elasticsearch/cluster/coordination/JoinValidationService.java
@@ -229,23 +229,22 @@ private void processNextItem() {
         try {
             nextItem.run();
         } finally {
-            var cleanupSuccess = false;
             try {
                 final var remaining = queueSize.decrementAndGet();
                 assert remaining >= 0;
                 if (remaining > 0) {
                     runProcessor();
                 }
-                cleanupSuccess = true;
-            } finally {
-                assert cleanupSuccess;
+            } catch (Exception e) {
+                assert false : e;
+                /* we only catch so we can assert false, so throwing is ok */
+                // noinspection ThrowFromFinallyBlock
+                throw e;
             }
         }
     }
 
     private void onShutdown() {
-        var success = false;
-
         try {
             // shutting down when enqueueing the next processor run which means there is no active processor so it's safe to clear out the
             // cache ...
@@ -259,10 +258,9 @@ private void onShutdown() {
                     nextItem.onFailure(new NodeClosedException(transportService.getLocalNode()));
                 }
             } while (queueSize.decrementAndGet() > 0);
-
-            success = true;
-        } finally {
-            assert success;
+        } catch (Exception e) {
+            assert false : e;
+            throw e;
         }
     }