Fix completeWith exception handling

ActionListener.completeWith would catch exceptions from
listener.onResponse and deliver them to lister.onFailure, essentially
double notifying the listener. Instead we now assert that listeners do
not throw when using ActionListener.completeWith.

Relates elastic#50886

server/src/main/java/org/elasticsearch/action/ActionListener.java

@@ -315,12 +315,28 @@ protected void innerOnFailure(Exception e) {
     /**
      * Completes the given listener with the result from the provided supplier accordingly.
      * This method is mainly used to complete a listener with a block of synchronous code.
+     *
+     * If the supplier fails, the listener's onFailure handler will be called.
+     * It is the responsibility of {@code delegate} to handle its own exceptions inside `onResponse` and `onFailure`.
      */
     static <Response> void completeWith(ActionListener<Response> listener, CheckedSupplier<Response, ? extends Exception> supplier) {
+        Response response;
         try {
-            listener.onResponse(supplier.get());
+            response = supplier.get();
         } catch (Exception e) {
-            listener.onFailure(e);
+            try {
+                listener.onFailure(e);
+            } catch (RuntimeException ex) {
+                assert false : ex;
+                throw ex;
+            }
+            return;
+        }
+        try {
+            listener.onResponse(response);
+        } catch (RuntimeException ex) {
+            assert false : ex;
+            throw ex;
         }
     }
 }

server/src/test/java/org/elasticsearch/action/ActionListenerTests.java

@@ -233,6 +233,33 @@ public void testCompleteWith() {
         ActionListener.completeWith(onFailureListener, () -> { throw new IOException("not found"); });
         assertThat(onFailureListener.isDone(), equalTo(true));
         assertThat(expectThrows(ExecutionException.class, onFailureListener::get).getCause(), instanceOf(IOException.class));
+
+        AtomicReference<Exception> exReference = new AtomicReference<>();
+        ActionListener<String> listener = new ActionListener<>() {
+            @Override
+            public void onResponse(String s) {
+                if (s == null) {
+                    throw new IllegalArgumentException("simulate onResponse exception");
+                }
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                exReference.set(e);
+                if (e instanceof IllegalArgumentException) {
+                    throw (IllegalArgumentException) e;
+                }
+            }
+        };
+
+        AssertionError assertionError = expectThrows(AssertionError.class, () -> ActionListener.completeWith(listener, () -> null));
+        assertThat(assertionError.getCause(), instanceOf(IllegalArgumentException.class));
+        assertNull(exReference.get());
+
+        assertionError = expectThrows(AssertionError.class, () -> ActionListener.completeWith(listener,
+            () -> { throw new IllegalArgumentException(); }));
+        assertThat(assertionError.getCause(), instanceOf(IllegalArgumentException.class));
+        assertThat(exReference.get(), instanceOf(IllegalArgumentException.class));
     }
 
     /**