Ensure cleanups succeed in JoinValidationService (#90601) (#91253)

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.

Author: DaveCTurner

server/src/main/java/org/elasticsearch/cluster/coordination/JoinValidationService.java

@@ -229,27 +229,39 @@ private void processNextItem() {
         try {
             nextItem.run();
         } finally {
-            final var remaining = queueSize.decrementAndGet();
-            assert remaining >= 0;
-            if (remaining > 0) {
-                runProcessor();
+            try {
+                final var remaining = queueSize.decrementAndGet();
+                assert remaining >= 0;
+                if (remaining > 0) {
+                    runProcessor();
+                }
+            } 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() {
-        // 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);
+        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);
+        } catch (Exception e) {
+            assert false : e;
+            throw e;
+        }
     }
 
     private final AbstractRunnable cacheClearer = new AbstractRunnable() {