Fix completeWith exception handling (#51734)

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 #50886

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

@@ -315,12 +315,28 @@ public interface ActionListener<Response> {
     /**
      * 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) {
+            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 class ActionListenerTests extends ESTestCase {
         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<String>() {
+            @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));
     }
 
     /**