Fix ActionListener.map exception handling

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

@@ -136,14 +136,27 @@ static <Response> ActionListener<Response> wrap(Runnable runnable) {
      * Creates a listener that wraps another listener, mapping response values via the given mapping function and passing along
      * exceptions to the delegate.
      *
+     * Notice that if the listener onResponse handler fails, the exception will bubble out, whereas if the function fails, the listeners
+     * onFailure handler will be called. The principle is that the code using this is responsible for the function, whereas the listener
+     * should do its own exception handling since it is a different component.
+     *
      * @param listener Listener to delegate to
      * @param fn Function to apply to listener response
      * @param <Response> Response type of the new listener
      * @param <T> Response type of the wrapped listener
      * @return a listener that maps the received response and then passes it to its delegate listener
      */
     static <T, Response> ActionListener<Response> map(ActionListener<T> listener, CheckedFunction<Response, T, Exception> fn) {
-        return wrap(r -> listener.onResponse(fn.apply(r)), listener::onFailure);
+        return delegateFailure(listener, (ActionListener<T> delegate, Response response) -> {
+            T mapped;
+            try {
+                mapped = fn.apply(response);
+            } catch (Exception e) {
+                delegate.onFailure(e);
+                return;
+            }
+            delegate.onResponse(mapped);
+        });
     }
 
     /**

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

@@ -234,4 +234,38 @@ public void testCompleteWith() {
         assertThat(onFailureListener.isDone(), equalTo(true));
         assertThat(expectThrows(ExecutionException.class, onFailureListener::get).getCause(), instanceOf(IOException.class));
     }
+
+    public void testMap() {
+        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);
+            }
+        };
+        ActionListener<Boolean> mapped = ActionListener.map(listener, b -> {
+            if (b == null) {
+                return null;
+            } else if (b) {
+                throw new IllegalStateException("simulate map function exception");
+            } else {
+                return b.toString();
+            }
+        });
+
+        expectThrows(IllegalArgumentException.class, () -> mapped.onResponse(null));
+        assertNull(exReference.get());
+        mapped.onResponse(false);
+        assertNull(exReference.get());
+        mapped.onResponse(true);
+        assertThat(exReference.get(), instanceOf(IllegalStateException.class));
+    }
 }

server/src/test/java/org/elasticsearch/action/admin/indices/create/CreateIndexIT.java

@@ -19,6 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.create;
 
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.UnavailableShardsException;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
@@ -32,7 +35,9 @@
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.MappingMetaData;
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.service.ClusterApplierService;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
+import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.XContentFactory;
@@ -44,6 +49,7 @@
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
+import org.elasticsearch.test.MockLogAppender;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -379,4 +385,43 @@ public void testIndexNameInResponse() {
         assertEquals("Should have index name in response", "foo", response.index());
     }
 
+    public void testOnResponseFailureOnMaster() throws Exception {
+        final String exceptionMessage = randomAlphaOfLength(10);
+        final Logger logger = LogManager.getLogger(ClusterApplierService.class);
+        final MockLogAppender appender = new MockLogAppender();
+        appender.start();
+        appender.addExpectation(
+            new MockLogAppender.ExceptionSeenEventExpectation(
+                getTestName(),
+                logger.getName(),
+                Level.WARN,
+                "failed to notify ClusterStateListener",
+                TestException.class,
+                exceptionMessage));
+        try {
+            Loggers.addAppender(logger, appender);
+
+            internalCluster().masterClient().admin().indices().prepareCreate("test").execute(new ActionListener<>() {
+                @Override
+                public void onResponse(CreateIndexResponse response) {
+                    throw new TestException(exceptionMessage);
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    fail();
+                }
+            });
+            assertBusy(appender::assertAllExpectationsMatched);
+        } finally {
+            Loggers.removeAppender(logger, appender);
+            appender.stop();
+        }
+    }
+
+    private static class TestException extends RuntimeException {
+        TestException(String message) {
+            super(message);
+        }
+    }
 }