Call ActionListener.onResponse exactly once (#61584) (#61682)

Under specific circumstances we would call onResponse twice, which led to unexpected behavior.

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/esnative/ReservedRealm.java

@@ -129,8 +129,9 @@ public class ReservedRealm extends CachingUsernamePasswordRealm {
         if (realmEnabled == false) {
             if (anonymousEnabled && AnonymousUser.isAnonymousUsername(username, config.settings())) {
                 listener.onResponse(anonymousUser);
+            } else {
+                listener.onResponse(null);
             }
-            listener.onResponse(null);
         } else if (ClientReservedRealm.isReserved(username, config.settings()) == false) {
             listener.onResponse(null);
         } else if (AnonymousUser.isAnonymousUsername(username, config.settings())) {

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/esnative/ReservedRealmTests.java

@@ -42,11 +42,13 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Predicate;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.Matchers.any;
@@ -195,7 +197,7 @@ public class ReservedRealmTests extends ESTestCase {
         verifyVersionPredicate(principal, predicateCaptor.getValue());
 
         PlainActionFuture<User> future = new PlainActionFuture<>();
-        reservedRealm.doLookupUser("foobar", future);
+        reservedRealm.doLookupUser("foobar", assertListenerIsOnlyCalledOnce(future));
         final User doesntExist = future.actionGet();
         assertThat(doesntExist, nullValue());
         verifyNoMoreInteractions(usersStore);
@@ -210,12 +212,29 @@ public class ReservedRealmTests extends ESTestCase {
         final String principal = expectedUser.principal();
 
         PlainActionFuture<User> listener = new PlainActionFuture<>();
-        reservedRealm.doLookupUser(principal, listener);
+        reservedRealm.doLookupUser(principal, assertListenerIsOnlyCalledOnce(listener));
         final User user = listener.actionGet();
         assertNull(user);
         verifyZeroInteractions(usersStore);
     }
 
+
+    public void testLookupDisabledAnonymous() throws Exception {
+        Settings settings = Settings.builder()
+            .put(XPackSettings.RESERVED_REALM_ENABLED_SETTING.getKey(), false)
+            .put(AnonymousUser.ROLES_SETTING.getKey(), "anonymous")
+            .build();
+        final ReservedRealm reservedRealm =
+            new ReservedRealm(mock(Environment.class), settings, usersStore, new AnonymousUser(settings),
+                securityIndex, threadPool);
+        final User expectedUser = new AnonymousUser(settings);
+        final String principal = expectedUser.principal();
+
+        PlainActionFuture<User> listener = new PlainActionFuture<>();
+        reservedRealm.doLookupUser(principal, assertListenerIsOnlyCalledOnce(listener));
+        assertThat(listener.actionGet(), equalTo(expectedUser));
+    }
+
     public void testLookupThrows() throws Exception {
         final ReservedRealm reservedRealm =
             new ReservedRealm(mock(Environment.class), Settings.EMPTY, usersStore,
@@ -480,4 +499,13 @@ public class ReservedRealmTests extends ESTestCase {
         }
         assertThat(versionPredicate.test(Version.V_7_0_0), is(true));
     }
+
+    private static <T> ActionListener<T> assertListenerIsOnlyCalledOnce(ActionListener<T> delegate) {
+        final AtomicInteger callCount = new AtomicInteger(0);
+        return ActionListener.runBefore(delegate, () -> {
+            if (callCount.incrementAndGet() != 1) {
+                fail("Listener was called twice");
+            }
+        });
+    }
 }