Merge remote-tracking branch 'origin/master' into json_strict_duplicate_checks

Original commit: elastic/x-pack-elasticsearch@601a663982

elasticsearch/src/main/java/org/elasticsearch/xpack/graph/rest/action/RestGraphAction.java

@@ -5,37 +5,34 @@
  */
 package org.elasticsearch.xpack.graph.rest.action;
 
-import static org.elasticsearch.rest.RestRequest.Method.GET;
+import org.elasticsearch.ElasticsearchParseException;
-import static org.elasticsearch.rest.RestRequest.Method.POST;
+import org.elasticsearch.action.support.IndicesOptions;
-import static org.elasticsearch.xpack.graph.action.GraphExploreAction.INSTANCE;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.index.query.QueryParseContext;
+import org.elasticsearch.indices.query.IndicesQueriesRegistry;
+import org.elasticsearch.rest.RestController;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestToXContentListener;
+import org.elasticsearch.xpack.XPackClient;
+import org.elasticsearch.xpack.graph.action.GraphExploreRequest;
+import org.elasticsearch.xpack.graph.action.GraphExploreRequest.TermBoost;
+import org.elasticsearch.xpack.graph.action.Hop;
+import org.elasticsearch.xpack.graph.action.VertexRequest;
+import org.elasticsearch.xpack.rest.XPackRestHandler;
 
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 
-import org.elasticsearch.ElasticsearchParseException;
+import static org.elasticsearch.rest.RestRequest.Method.GET;
-import org.elasticsearch.action.support.IndicesOptions;
+import static org.elasticsearch.rest.RestRequest.Method.POST;
-import org.elasticsearch.common.ParseField;
+import static org.elasticsearch.xpack.graph.action.GraphExploreAction.INSTANCE;
-import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.xcontent.XContentFactory;
-import org.elasticsearch.common.xcontent.XContentParser;
-import org.elasticsearch.index.query.QueryParseContext;
-import org.elasticsearch.indices.query.IndicesQueriesRegistry;
-import org.elasticsearch.rest.RestController;
-import org.elasticsearch.rest.RestRequest;
-import org.elasticsearch.rest.action.RestActions;
-import org.elasticsearch.rest.action.RestToXContentListener;
-import org.elasticsearch.xpack.XPackClient;
-import org.elasticsearch.xpack.graph.action.GraphExploreRequest;
-import org.elasticsearch.xpack.graph.action.Hop;
-import org.elasticsearch.xpack.graph.action.VertexRequest;
-import org.elasticsearch.xpack.graph.action.GraphExploreRequest.TermBoost;
-import org.elasticsearch.xpack.rest.XPackRestHandler;
 
 /**
  * @see GraphExploreRequest
@@ -88,14 +85,13 @@ public class RestGraphAction extends XPackRestHandler {
         if (request.hasParam(TIMEOUT_FIELD.getPreferredName())) {
             graphRequest.timeout(request.paramAsTime(TIMEOUT_FIELD.getPreferredName(), null));
         }
-        if (!RestActions.hasBodyContent(request)) {
+        if (false == request.hasContentOrSourceParam()) {
             throw new ElasticsearchParseException("Body missing for graph request");
         }
-        BytesReference qBytes = RestActions.getRestContent(request);
 
         Hop currentHop = graphRequest.createNextHop(null);
 
-        try(XContentParser parser = XContentFactory.xContent(qBytes).createParser(qBytes)) {
+        try (XContentParser parser = request.contentOrSourceParamParser()) {
             QueryParseContext context = new QueryParseContext(indicesQueriesRegistry, parser, parseFieldMatcher);
 
             XContentParser.Token token = parser.nextToken();

elasticsearch/src/main/java/org/elasticsearch/xpack/monitoring/rest/action/RestMonitoringBulkAction.java

@@ -59,7 +59,7 @@ public class RestMonitoringBulkAction extends MonitoringRestHandler {
             throw new IllegalArgumentException("no [" + INTERVAL + "] for monitoring bulk request");
         }
 
-        if (!RestActions.hasBodyContent(request)) {
+        if (false == request.hasContentOrSourceParam()) {
             throw new ElasticsearchParseException("no body content for monitoring bulk request");
         }
 

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/AuthenticationService.java

@@ -343,8 +343,7 @@ public class AuthenticationService extends AbstractComponent {
          */
         private void lookupRunAsUser(final User user, String runAsUsername, Consumer<User> userConsumer) {
             final List<Realm> realmsList = realms.asList();
-            final BiConsumer<Realm, ActionListener<User>> realmLookupConsumer = (realm, lookupUserListener) -> {
+            final BiConsumer<Realm, ActionListener<User>> realmLookupConsumer = (realm, lookupUserListener) ->
-                if (realm.userLookupSupported()) {
                     realm.lookupUser(runAsUsername, ActionListener.wrap((lookedupUser) -> {
                         if (lookedupUser != null) {
                             lookedupBy = new RealmRef(realm.name(), realm.type(), nodeName);
@@ -353,10 +352,6 @@ public class AuthenticationService extends AbstractComponent {
                             lookupUserListener.onResponse(null);
                         }
                     }, lookupUserListener::onFailure));
-                } else {
-                    lookupUserListener.onResponse(null);
-                }
-            };
 
             final IteratingActionListener<User, Realm> userLookupListener =
                     new IteratingActionListener<>(ActionListener.wrap((lookupUser) -> userConsumer.accept(new User(user, lookupUser)),

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/Realm.java

@@ -71,44 +71,24 @@ public abstract class Realm implements Comparable<Realm> {
     public abstract AuthenticationToken token(ThreadContext context);
 
     /**
-     * Authenticates the given token in a blocking fashion. A successful authentication will return the User associated
+     * Authenticates the given token in an asynchronous fashion. A successful authentication will call the
-     * with the given token. An unsuccessful authentication returns {@code null}. This method is deprecated in favor of
+     * {@link ActionListener#onResponse} with the User associated with the given token. An unsuccessful authentication calls
-     * {@link #authenticate(AuthenticationToken, ActionListener)}.
+     * with {@code null} on the argument.
      *
      * @param token The authentication token
-     * @return      The authenticated user or {@code null} if authentication failed.
+     * @param listener The listener to pass the authentication result to
-     *
      */
-    @Deprecated
+    public abstract void authenticate(AuthenticationToken token, ActionListener<User> listener);
-    public abstract User authenticate(AuthenticationToken token);
-
-    public void authenticate(AuthenticationToken token, ActionListener<User> listener) {
-        try {
-            listener.onResponse(authenticate(token));
-        } catch (Exception e) {
-            listener.onFailure(e);
-        }
-    }
 
     /**
-     * Looks up the user identified the String identifier. A successful lookup will return the {@link User} identified
+     * Looks up the user identified the String identifier. A successful lookup will call the {@link ActionListener#onResponse}
-     * by the username. An unsuccessful lookup returns {@code null}. This method is deprecated in favor of
+     * with the {@link User} identified by the username. An unsuccessful lookup call with {@code null} as the argument. If lookup is not
-     * {@link #lookupUser(String, ActionListener)}
+     * supported, simply return {@code null} when called.
      *
      * @param username the String identifier for the user
-     * @return         the {@link User} or {@code null} if lookup failed
+     * @param listener The listener to pass the lookup result to
      */
-    @Deprecated
+    public abstract void lookupUser(String username, ActionListener<User> listener);
-    public abstract User lookupUser(String username);
-
-    public void lookupUser(String username, ActionListener<User> listener) {
-        try {
-            User user = lookupUser(username);
-            listener.onResponse(user);
-        } catch (Exception e) {
-            listener.onFailure(e);
-        }
-    }
 
     public Map<String, Object> usageStats() {
         Map<String, Object> stats = new HashMap<>();
@@ -117,14 +97,6 @@ public abstract class Realm implements Comparable<Realm> {
         return stats;
     }
 
-    /**
-     * Indicates whether this realm supports user lookup. This method is deprecated. In the future if lookup is not supported, simply
-     * return null when called.
-     * @return true if the realm supports user lookup
-     */
-    @Deprecated
-    public abstract boolean userLookupSupported();
-
     @Override
     public String toString() {
         return type + "/" + config.name;

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/esnative/NativeRealm.java

@@ -25,11 +25,6 @@ public class NativeRealm extends CachingUsernamePasswordRealm {
         this.userStore = usersStore;
     }
 
-    @Override
-    public boolean userLookupSupported() {
-        return true;
-    }
-
     @Override
     protected void doLookupUser(String username, ActionListener<User> listener) {
         userStore.getUser(username, listener);

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/esnative/ReservedRealm.java

@@ -108,11 +108,6 @@ public class ReservedRealm extends CachingUsernamePasswordRealm {
         }
     }
 
-    @Override
-    public boolean userLookupSupported() {
-        return true;
-    }
-
     public static boolean isReserved(String username, Settings settings) {
         assert username != null;
         switch (username) {

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/file/FileRealm.java

@@ -18,8 +18,8 @@ public class FileRealm extends CachingUsernamePasswordRealm {
 
     public static final String TYPE = "file";
 
-    final FileUserPasswdStore userPasswdStore;
+    private final FileUserPasswdStore userPasswdStore;
-    final FileUserRolesStore userRolesStore;
+    private final FileUserRolesStore userRolesStore;
 
     public FileRealm(RealmConfig config, ResourceWatcherService watcherService) {
         this(config, new FileUserPasswdStore(config, watcherService), new FileUserRolesStore(config, watcherService));
@@ -61,9 +61,4 @@ public class FileRealm extends CachingUsernamePasswordRealm {
         stats.put("size", userPasswdStore.usersCount());
         return stats;
     }
-
-    @Override
-    public boolean userLookupSupported() {
-        return true;
-    }
 }

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/ldap/LdapRealm.java

@@ -101,11 +101,6 @@ public final class LdapRealm extends CachingUsernamePasswordRealm {
         }
     }
 
-    @Override
-    public boolean userLookupSupported() {
-        return sessionFactory.supportsUnauthenticatedSession();
-    }
-
     @Override
     public Map<String, Object> usageStats() {
         Map<String, Object> usage = super.usageStats();

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/pki/PkiRealm.java

@@ -80,11 +80,6 @@ public class PkiRealm extends Realm {
         return token(context.getTransient(PKI_CERT_HEADER_NAME), principalPattern, logger);
     }
 
-    @Override
-    public User authenticate(AuthenticationToken authToken) {
-        throw new UnsupportedOperationException("internal realms do not support blocking calls");
-    }
-
     @Override
     public void authenticate(AuthenticationToken authToken, ActionListener<User> listener) {
         X509AuthenticationToken token = (X509AuthenticationToken)authToken;
@@ -97,13 +92,8 @@ public class PkiRealm extends Realm {
     }
 
     @Override
-    public User lookupUser(String username) {
+    public void lookupUser(String username, ActionListener<User> listener) {
-        return null;
+        listener.onResponse(null);
-    }
-
-    @Override
-    public boolean userLookupSupported() {
-        return false;
     }
 
     static X509AuthenticationToken token(Object pkiHeaderValue, Pattern principalPattern, Logger logger) {

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/support/CachingUsernamePasswordRealm.java

@@ -140,9 +140,7 @@ public abstract class CachingUsernamePasswordRealm extends UsernamePasswordRealm
 
     @Override
     public final void lookupUser(String username, ActionListener<User> listener) {
-        if (!userLookupSupported()) {
+        if (cache != null) {
-            listener.onResponse(null);
-        } else if (cache != null) {
             UserWithHash withHash = cache.get(username);
             if (withHash == null) {
                 try {

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/support/UsernamePasswordRealm.java

@@ -25,12 +25,4 @@ abstract class UsernamePasswordRealm extends Realm {
     public boolean supports(AuthenticationToken token) {
         return token instanceof UsernamePasswordToken;
     }
-
-    public final User authenticate(AuthenticationToken token) {
-        throw new UnsupportedOperationException("internal realms should not support blocking calls!!");
-    }
-
-    public final User lookupUser(String username) {
-        throw new UnsupportedOperationException("internal realms should not support blocking calls!");
-    }
 }

elasticsearch/src/main/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptor.java

@@ -8,11 +8,13 @@ package org.elasticsearch.xpack.security.transport;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.DestructiveOperations;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.CheckedConsumer;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportChannel;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportInterceptor;
@@ -78,25 +80,25 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
     public AsyncSender interceptSender(AsyncSender sender) {
         return new AsyncSender() {
             @Override
-            public <T extends TransportResponse> void sendRequest(DiscoveryNode node, String action, TransportRequest request,
+            public <T extends TransportResponse> void sendRequest(Transport.Connection connection, String action, TransportRequest request,
                                                                   TransportRequestOptions options, TransportResponseHandler<T> handler) {
                 if (licenseState.isAuthAllowed()) {
                     // Sometimes a system action gets executed like a internal create index request or update mappings request
                     // which means that the user is copied over to system actions so we need to change the user
                     if (AuthorizationUtils.shouldReplaceUserWithSystem(threadPool.getThreadContext(), action)) {
-                        securityContext.executeAsUser(SystemUser.INSTANCE, (original) -> sendWithUser(node, action, request, options,
+                        securityContext.executeAsUser(SystemUser.INSTANCE, (original) -> sendWithUser(connection, action, request, options,
                                 new ContextRestoreResponseHandler<>(threadPool.getThreadContext(), original, handler), sender));
                     } else {
-                        sendWithUser(node, action, request, options, handler, sender);
+                        sendWithUser(connection, action, request, options, handler, sender);
                     }
                 } else {
-                    sender.sendRequest(node, action, request, options, handler);
+                    sender.sendRequest(connection, action, request, options, handler);
                 }
             }
         };
     }
 
-    private <T extends TransportResponse> void sendWithUser(DiscoveryNode node, String action, TransportRequest request,
+    private <T extends TransportResponse> void sendWithUser(Transport.Connection connection, String action, TransportRequest request,
                                                             TransportRequestOptions options, TransportResponseHandler<T> handler,
                                                             AsyncSender sender) {
         // There cannot be a request outgoing from this node that is not associated with a user.
@@ -105,7 +107,7 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
         }
 
         try {
-            sender.sendRequest(node, action, request, options, handler);
+            sender.sendRequest(connection, action, request, options, handler);
         } catch (Exception e) {
             handler.handleException(new TransportException("failed sending request", e));
         }
@@ -213,7 +215,7 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
                     assert filter != null;
                     final Thread executingThread = Thread.currentThread();
 
-                    ActionListener.CheckedConsumer<Void> consumer = (x) -> {
+                    CheckedConsumer<Void, Exception> consumer = (x) -> {
                         final Executor executor;
                         if (executingThread == Thread.currentThread()) {
                             // only fork off if we get called on another thread this means we moved to

elasticsearch/src/main/java/org/elasticsearch/xpack/watcher/execution/ExecutionService.java

@@ -34,6 +34,7 @@ import org.joda.time.DateTime;
 
 import java.time.Clock;
 import java.util.ArrayList;
+import java.util.BitSet;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -187,16 +188,13 @@ public class ExecutionService extends AbstractComponent {
 
         logger.debug("saving watch records [{}]", triggeredWatches.size());
 
-        triggeredWatchStore.putAll(triggeredWatches, new ActionListener<List<Integer>>() {
+        triggeredWatchStore.putAll(triggeredWatches, new ActionListener<BitSet>() {
             @Override
-            public void onResponse(List<Integer> successFullSlots) {
+            public void onResponse(BitSet slots) {
-                for (Integer slot : successFullSlots) {
+                int slot = 0;
-                    TriggeredWatch triggeredWatch = triggeredWatches.get(slot);
+                while ((slot = slots.nextSetBit(slot)) != -1) {
-                    try {
+                    executeAsync(contexts.get(slot), triggeredWatches.get(slot));
-                        executeAsync(contexts.get(slot), triggeredWatch);
+                    slot++;
-                    } catch (Exception e) {
-                        logger.error((Supplier<?>) () -> new ParameterizedMessage("failed to execute watch [{}]", triggeredWatch.id()), e);
-                    }
                 }
             }
 
@@ -236,9 +234,11 @@ public class ExecutionService extends AbstractComponent {
             return;
         }
 
-        List<Integer> slots = triggeredWatchStore.putAll(triggeredWatches);
+        BitSet slots = triggeredWatchStore.putAll(triggeredWatches);
-        for (Integer slot : slots) {
+        int slot = 0;
+        while ((slot = slots.nextSetBit(slot)) != -1) {
             executeAsync(contexts.get(slot), triggeredWatches.get(slot));
+            slot++;
         }
     }
 

elasticsearch/src/main/java/org/elasticsearch/xpack/watcher/execution/TriggeredWatchStore.java

@@ -33,6 +33,7 @@ import org.elasticsearch.xpack.watcher.watch.WatchStoreUtils;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.BitSet;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -129,10 +130,10 @@ public class TriggeredWatchStore extends AbstractComponent {
         }
     }
 
-    public void putAll(final List<TriggeredWatch> triggeredWatches, final ActionListener<List<Integer>> listener) throws Exception {
+    public void putAll(final List<TriggeredWatch> triggeredWatches, final ActionListener<BitSet> listener) throws Exception {
 
         if (triggeredWatches.isEmpty()) {
-            listener.onResponse(Collections.emptyList());
+            listener.onResponse(new BitSet(0));
             return;
         }
 
@@ -140,7 +141,9 @@ public class TriggeredWatchStore extends AbstractComponent {
             put(triggeredWatches.get(0), new ActionListener<Boolean>() {
                 @Override
                 public void onResponse(Boolean success) {
-                    listener.onResponse(Collections.singletonList(0));
+                    BitSet bitSet = new BitSet(1);
+                    bitSet.set(0);
+                    listener.onResponse(bitSet);
                 }
 
                 @Override
@@ -163,15 +166,14 @@ public class TriggeredWatchStore extends AbstractComponent {
             client.bulk(request, new ActionListener<BulkResponse>() {
                 @Override
                 public void onResponse(BulkResponse response) {
-                    List<Integer> successFullSlots = new ArrayList<Integer>();
+                    BitSet successFullSlots = new BitSet(triggeredWatches.size());
                     for (int i = 0; i < response.getItems().length; i++) {
                         BulkItemResponse itemResponse = response.getItems()[i];
                         if (itemResponse.isFailed()) {
                             logger.error("could store triggered watch with id [{}], because failed [{}]", itemResponse.getId(),
                                     itemResponse.getFailureMessage());
                         } else {
-                            IndexResponse indexResponse = itemResponse.getResponse();
+                            successFullSlots.set(i);
-                            successFullSlots.add(i);
                         }
                     }
                     listener.onResponse(successFullSlots);
@@ -187,7 +189,7 @@ public class TriggeredWatchStore extends AbstractComponent {
         }
     }
 
-    public List<Integer> putAll(final List<TriggeredWatch> triggeredWatches) throws Exception {
+    public BitSet putAll(final List<TriggeredWatch> triggeredWatches) throws Exception {
         ensureStarted();
         try {
             BulkRequest request = new BulkRequest();
@@ -198,15 +200,14 @@ public class TriggeredWatchStore extends AbstractComponent {
                 request.add(indexRequest);
             }
             BulkResponse response = client.bulk(request, (TimeValue) null);
-            List<Integer> successFullSlots = new ArrayList<>();
+            BitSet successFullSlots = new BitSet(triggeredWatches.size());
             for (int i = 0; i < response.getItems().length; i++) {
                 BulkItemResponse itemResponse = response.getItems()[i];
                 if (itemResponse.isFailed()) {
                     logger.error("could store triggered watch with id [{}], because failed [{}]", itemResponse.getId(),
                             itemResponse.getFailureMessage());
                 } else {
-                    IndexResponse indexResponse = itemResponse.getResponse();
+                    successFullSlots.set(i);
-                    successFullSlots.add(i);
                 }
             }
             return successFullSlots;

elasticsearch/src/test/java/org/elasticsearch/xpack/security/authc/AuthenticationServiceTests.java

@@ -172,7 +172,7 @@ public class AuthenticationServiceTests extends ESTestCase {
         assertThat(result.getUser(), is(user));
         verify(auditTrail).authenticationSuccess(secondRealm.name(), user, "_action", message);
         verifyNoMoreInteractions(auditTrail);
-        verify(firstRealm, never()).authenticate(token);
+        verify(firstRealm, never()).authenticate(eq(token), any(ActionListener.class));
         assertThreadContextContainsAuthentication(result);
     }
 
@@ -600,9 +600,9 @@ public class AuthenticationServiceTests extends ESTestCase {
         when(secondRealm.token(threadContext)).thenReturn(token);
         when(secondRealm.supports(token)).thenReturn(true);
         mockAuthenticate(secondRealm, token, new User("lookup user", new String[]{"user"}));
+        mockRealmLookupReturnsNull(firstRealm, "run_as");
         doThrow(authenticationError("realm doesn't want to lookup"))
             .when(secondRealm).lookupUser(eq("run_as"), any(ActionListener.class));
-        when(secondRealm.userLookupSupported()).thenReturn(true);
 
         try {
             authenticateBlocking("_action", message, null);
@@ -619,9 +619,9 @@ public class AuthenticationServiceTests extends ESTestCase {
         when(secondRealm.token(threadContext)).thenReturn(token);
         when(secondRealm.supports(token)).thenReturn(true);
         mockAuthenticate(secondRealm, token, new User("lookup user", new String[]{"user"}));
+        mockRealmLookupReturnsNull(firstRealm, "run_as");
         doThrow(authenticationError("realm doesn't want to " + "lookup"))
                 .when(secondRealm).lookupUser(eq("run_as"), any(ActionListener.class));
-        when(secondRealm.userLookupSupported()).thenReturn(true);
 
         try {
             authenticateBlocking(restRequest);
@@ -640,12 +640,12 @@ public class AuthenticationServiceTests extends ESTestCase {
         final User user = new User("lookup user", new String[]{"user"}, "lookup user", "lookup@foo.foo",
                 Collections.singletonMap("foo", "bar"), true);
         mockAuthenticate(secondRealm, token, user);
+        mockRealmLookupReturnsNull(firstRealm, "run_as");
         doAnswer((i) -> {
             ActionListener listener = (ActionListener) i.getArguments()[1];
             listener.onResponse(new User("looked up user", new String[]{"some role"}));
             return null;
         }).when(secondRealm).lookupUser(eq("run_as"), any(ActionListener.class));
-        when(secondRealm.userLookupSupported()).thenReturn(true);
 
         Authentication result;
         if (randomBoolean()) {
@@ -676,13 +676,11 @@ public class AuthenticationServiceTests extends ESTestCase {
         when(secondRealm.token(threadContext)).thenReturn(token);
         when(secondRealm.supports(token)).thenReturn(true);
         mockAuthenticate(secondRealm, token, new User("lookup user", new String[]{"user"}));
-        when(firstRealm.userLookupSupported()).thenReturn(true);
         doAnswer((i) -> {
             ActionListener listener = (ActionListener) i.getArguments()[1];
             listener.onResponse(new User("looked up user", new String[]{"some role"}));
             return null;
         }).when(firstRealm).lookupUser(eq("run_as"), any(ActionListener.class));
-        when(firstRealm.userLookupSupported()).thenReturn(true);
 
         Authentication result;
         if (randomBoolean()) {
@@ -709,7 +707,6 @@ public class AuthenticationServiceTests extends ESTestCase {
         when(secondRealm.token(threadContext)).thenReturn(token);
         when(secondRealm.supports(token)).thenReturn(true);
         mockAuthenticate(secondRealm, token, user);
-        when(secondRealm.userLookupSupported()).thenReturn(true);
 
         try {
             authenticateBlocking(restRequest);
@@ -727,7 +724,6 @@ public class AuthenticationServiceTests extends ESTestCase {
         when(secondRealm.token(threadContext)).thenReturn(token);
         when(secondRealm.supports(token)).thenReturn(true);
         mockAuthenticate(secondRealm, token, user);
-        when(secondRealm.userLookupSupported()).thenReturn(true);
 
         try {
             authenticateBlocking("_action", message, null);
@@ -744,12 +740,12 @@ public class AuthenticationServiceTests extends ESTestCase {
         when(secondRealm.token(threadContext)).thenReturn(token);
         when(secondRealm.supports(token)).thenReturn(true);
         mockAuthenticate(secondRealm, token, new User("lookup user", new String[]{"user"}));
+        mockRealmLookupReturnsNull(firstRealm, "run_as");
         doAnswer((i) -> {
             ActionListener listener = (ActionListener) i.getArguments()[1];
             listener.onResponse(new User("looked up user", new String[]{"some role"}, null, null, null, false));
             return null;
         }).when(secondRealm).lookupUser(eq("run_as"), any(ActionListener.class));
-        when(secondRealm.userLookupSupported()).thenReturn(true);
         User fallback = randomBoolean() ? SystemUser.INSTANCE : null;
         ElasticsearchSecurityException e =
                 expectThrows(ElasticsearchSecurityException.class, () -> authenticateBlocking("_action", message, fallback));
@@ -764,12 +760,12 @@ public class AuthenticationServiceTests extends ESTestCase {
         when(secondRealm.token(threadContext)).thenReturn(token);
         when(secondRealm.supports(token)).thenReturn(true);
         mockAuthenticate(secondRealm, token, new User("lookup user", new String[]{"user"}));
+        mockRealmLookupReturnsNull(firstRealm, "run_as");
         doAnswer((i) -> {
             ActionListener listener = (ActionListener) i.getArguments()[1];
             listener.onResponse(new User("looked up user", new String[]{"some role"}, null, null, null, false));
             return null;
         }).when(secondRealm).lookupUser(eq("run_as"), any(ActionListener.class));
-        when(secondRealm.userLookupSupported()).thenReturn(true);
 
         ElasticsearchSecurityException e =
                 expectThrows(ElasticsearchSecurityException.class, () -> authenticateBlocking(restRequest));
@@ -816,6 +812,14 @@ public class AuthenticationServiceTests extends ESTestCase {
         return future.actionGet();
     }
 
+    private static void mockRealmLookupReturnsNull(Realm realm, String username) {
+        doAnswer((i) -> {
+            ActionListener listener = (ActionListener) i.getArguments()[1];
+            listener.onResponse(null);
+            return null;
+        }).when(realm).lookupUser(eq(username), any(ActionListener.class));
+    }
+
     static class TestRealms extends Realms {
 
         TestRealms(Settings settings, Environment env, Map<String, Factory> factories, XPackLicenseState licenseState,

elasticsearch/src/test/java/org/elasticsearch/xpack/security/authc/RealmsTests.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.security.authc;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.env.Environment;
@@ -433,18 +434,13 @@ public class RealmsTests extends ESTestCase {
         }
 
         @Override
-        public User authenticate(AuthenticationToken token) {
+        public void authenticate(AuthenticationToken token, ActionListener<User> listener) {
-            return null;
+            listener.onResponse(null);
         }
 
         @Override
-        public User lookupUser(String username) {
+        public void lookupUser(String username, ActionListener<User> listener) {
-            return null;
+            listener.onResponse(null);
-        }
-
-        @Override
-        public boolean userLookupSupported() {
-            return false;
         }
     }
 }

elasticsearch/src/test/java/org/elasticsearch/xpack/security/authc/support/CachingUsernamePasswordRealmTests.java

@@ -58,11 +58,6 @@ public class CachingUsernamePasswordRealmTests extends ESTestCase {
             protected void doLookupUser(String username, ActionListener<User> listener) {
                 listener.onFailure(new UnsupportedOperationException("this method should not be called"));
             }
-
-            @Override
-            public boolean userLookupSupported() {
-                return false;
-            }
         };
 
         assertThat(realm.hasher, sameInstance(Hasher.resolve(hashAlgo)));
@@ -213,26 +208,6 @@ public class CachingUsernamePasswordRealmTests extends ESTestCase {
         assertThat(e.getMessage() , containsString("lookup exception"));
     }
 
-    public void testThatLookupIsNotCalledIfNotSupported() throws Exception {
-        LookupNotSupportedRealm realm = new LookupNotSupportedRealm(globalSettings);
-        assertThat(realm.userLookupSupported(), is(false));
-        PlainActionFuture<User> future = new PlainActionFuture<>();
-        realm.lookupUser("a", future);
-        User user = future.actionGet();
-        assertThat(user, is(nullValue()));
-        assertThat(realm.lookupInvocationCounter.intValue(), is(0));
-
-        // try to lookup more
-        future = new PlainActionFuture<>();
-        realm.lookupUser("b", future);
-        future.actionGet();
-        future = new PlainActionFuture<>();
-        realm.lookupUser("c", future);
-        future.actionGet();
-
-        assertThat(realm.lookupInvocationCounter.intValue(), is(0));
-    }
-
     public void testCacheConcurrency() throws Exception {
         final String username = "username";
         final SecuredString password = new SecuredString("changeme".toCharArray());
@@ -255,11 +230,6 @@ public class CachingUsernamePasswordRealmTests extends ESTestCase {
             protected void doLookupUser(String username, ActionListener<User> listener) {
                 listener.onFailure(new UnsupportedOperationException("this method should not be called"));
             }
-
-            @Override
-            public boolean userLookupSupported() {
-                return false;
-            }
         };
 
         final CountDownLatch latch = new CountDownLatch(1);
@@ -317,11 +287,6 @@ public class CachingUsernamePasswordRealmTests extends ESTestCase {
             protected void doLookupUser(String username, ActionListener<User> listener) {
                 listener.onResponse(new User(username, new String[]{"r1", "r2", "r3"}));
             }
-
-            @Override
-            public boolean userLookupSupported() {
-                return true;
-            }
         };
 
         final CountDownLatch latch = new CountDownLatch(1);
@@ -366,11 +331,6 @@ public class CachingUsernamePasswordRealmTests extends ESTestCase {
             super("failing", new RealmConfig("failing-test", settings, global));
         }
 
-        @Override
-        public boolean userLookupSupported() {
-            return true;
-        }
-
         @Override
         protected void doAuthenticate(UsernamePasswordToken token, ActionListener<User> listener) {
             listener.onResponse(null);
@@ -397,11 +357,6 @@ public class CachingUsernamePasswordRealmTests extends ESTestCase {
         protected void doLookupUser(String username, ActionListener<User> listener) {
             listener.onFailure(new RuntimeException("lookup exception"));
         }
-
-        @Override
-        public boolean userLookupSupported() {
-            return true;
-        }
     }
 
     static class AlwaysAuthenticateCachingRealm extends CachingUsernamePasswordRealm {
@@ -424,11 +379,6 @@ public class CachingUsernamePasswordRealmTests extends ESTestCase {
             lookupInvocationCounter.incrementAndGet();
             listener.onResponse(new User(username, new String[] { "lookupRole1", "lookupRole2" }));
         }
-
-        @Override
-        public boolean userLookupSupported() {
-            return true;
-        }
     }
 
     static class LookupNotSupportedRealm extends CachingUsernamePasswordRealm {
@@ -451,10 +401,5 @@ public class CachingUsernamePasswordRealmTests extends ESTestCase {
             lookupInvocationCounter.incrementAndGet();
             listener.onFailure(new UnsupportedOperationException("don't call lookup if lookup isn't supported!!!"));
         }
-
-        @Override
-        public boolean userLookupSupported() {
-            return false;
-        }
     }
 }

elasticsearch/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java

@@ -6,7 +6,6 @@
 package org.elasticsearch.xpack.security.transport;
 
 import org.elasticsearch.action.support.DestructiveOperations;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
@@ -14,6 +13,7 @@ import org.elasticsearch.env.Environment;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportInterceptor.AsyncSender;
 import org.elasticsearch.transport.TransportRequest;
@@ -77,7 +77,7 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
         AtomicBoolean calledWrappedSender = new AtomicBoolean(false);
         AsyncSender sender = interceptor.interceptSender(new AsyncSender() {
             @Override
-            public <T extends TransportResponse> void sendRequest(DiscoveryNode node, String action, TransportRequest request,
+            public <T extends TransportResponse> void sendRequest(Transport.Connection connection, String action, TransportRequest request,
                                                                   TransportRequestOptions options, TransportResponseHandler<T> handler) {
                 if (calledWrappedSender.compareAndSet(false, true) == false) {
                     fail("sender called more than once!");
@@ -104,7 +104,7 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
         AtomicReference<User> sendingUser = new AtomicReference<>();
         AsyncSender sender = interceptor.interceptSender(new AsyncSender() {
             @Override
-            public <T extends TransportResponse> void sendRequest(DiscoveryNode node, String action, TransportRequest request,
+            public <T extends TransportResponse> void sendRequest(Transport.Connection connection, String action, TransportRequest request,
                                                                   TransportRequestOptions options, TransportResponseHandler<T> handler) {
                 if (calledWrappedSender.compareAndSet(false, true) == false) {
                     fail("sender called more than once!");
@@ -136,7 +136,7 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
         AtomicReference<User> sendingUser = new AtomicReference<>();
         AsyncSender sender = interceptor.interceptSender(new AsyncSender() {
             @Override
-            public <T extends TransportResponse> void sendRequest(DiscoveryNode node, String action, TransportRequest request,
+            public <T extends TransportResponse> void sendRequest(Transport.Connection connection, String action, TransportRequest request,
                                                                   TransportRequestOptions options, TransportResponseHandler<T> handler) {
                 if (calledWrappedSender.compareAndSet(false, true) == false) {
                     fail("sender called more than once!");
@@ -163,7 +163,7 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
         assertNull(securityContext.getUser());
         AsyncSender sender = interceptor.interceptSender(new AsyncSender() {
             @Override
-            public <T extends TransportResponse> void sendRequest(DiscoveryNode node, String action, TransportRequest request,
+            public <T extends TransportResponse> void sendRequest(Transport.Connection connection, String action, TransportRequest request,
                                                                   TransportRequestOptions options, TransportResponseHandler<T> handler) {
                 fail("sender should not be called!");
             }

qa/security-example-realm/src/main/java/org/elasticsearch/example/realm/CustomRealm.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.example.realm;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.xpack.security.user.User;
 import org.elasticsearch.xpack.security.authc.AuthenticationToken;
@@ -46,22 +47,18 @@ public class CustomRealm extends Realm {
     }
 
     @Override
-    public User authenticate(AuthenticationToken authToken) {
+    public void authenticate(AuthenticationToken authToken, ActionListener<User> listener) {
         UsernamePasswordToken token = (UsernamePasswordToken)authToken;
         final String actualUser = token.principal();
         if (KNOWN_USER.equals(actualUser) && SecuredString.constantTimeEquals(token.credentials(), KNOWN_PW)) {
-            return new User(actualUser, ROLES);
+            listener.onResponse(new User(actualUser, ROLES));
+        } else {
+            listener.onResponse(null);
         }
-        return null;
     }
 
     @Override
-    public User lookupUser(String username) {
+    public void lookupUser(String username, ActionListener<User> listener) {
-        return null;
+        listener.onResponse(null);
-    }
-
-    @Override
-    public boolean userLookupSupported() {
-        return false;
     }
 }

qa/security-example-realm/src/test/java/org/elasticsearch/example/realm/CustomRealmTests.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.example.realm;
 
+import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.xpack.security.user.User;
 import org.elasticsearch.xpack.security.authc.RealmConfig;
@@ -22,7 +23,9 @@ public class CustomRealmTests extends ESTestCase {
         CustomRealm realm = new CustomRealm(new RealmConfig("test", Settings.EMPTY, globalSettings));
         SecuredString password = new SecuredString(CustomRealm.KNOWN_PW.toCharArray());
         UsernamePasswordToken token = new UsernamePasswordToken(CustomRealm.KNOWN_USER, password);
-        User user = realm.authenticate(token);
+        PlainActionFuture<User> plainActionFuture = new PlainActionFuture<>();
+        realm.authenticate(token, plainActionFuture);
+        User user = plainActionFuture.actionGet();
         assertThat(user, notNullValue());
         assertThat(user.roles(), equalTo(CustomRealm.ROLES));
         assertThat(user.principal(), equalTo(CustomRealm.KNOWN_USER));
@@ -33,7 +36,8 @@ public class CustomRealmTests extends ESTestCase {
         CustomRealm realm = new CustomRealm(new RealmConfig("test", Settings.EMPTY, globalSettings));
         SecuredString password = new SecuredString(CustomRealm.KNOWN_PW.toCharArray());
         UsernamePasswordToken token = new UsernamePasswordToken(CustomRealm.KNOWN_USER + "1", password);
-        User user = realm.authenticate(token);
+        PlainActionFuture<User> plainActionFuture = new PlainActionFuture<>();
-        assertThat(user, nullValue());
+        realm.authenticate(token, plainActionFuture);
+        assertThat(plainActionFuture.actionGet(), nullValue());
     }
 }