Do not refresh realm cache unless required

If there are no realms that depend on the native role mapping store,
then changes should it should not perform any cache refresh.
A refresh with an empty realm array will refresh all realms.

This also fixes a spurious log warning that could occur if the
role mapping store was notified that the security index was recovered
before any realm were attached.

Resolves: elastic#35218
Backport of: elastic#42169

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/support/mapper/NativeRoleMappingStore.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.action.support.ContextPreservingActionListener;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.CheckedBiConsumer;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
@@ -329,7 +330,12 @@ public void onSecurityIndexStateChange(SecurityIndexManager.State previousState,
     }
 
     private <Result> void refreshRealms(ActionListener<Result> listener, Result result) {
-        String[] realmNames = this.realmsToRefresh.toArray(new String[realmsToRefresh.size()]);
+        if (realmsToRefresh.isEmpty()) {
+            listener.onResponse(result);
+            return;
+        }
+
+        final String[] realmNames = this.realmsToRefresh.toArray(Strings.EMPTY_ARRAY);
         final SecurityClient securityClient = new SecurityClient(client);
         executeAsyncWithOrigin(client.threadPool().getThreadContext(), SECURITY_ORIGIN,
                 securityClient.prepareClearRealmCache().realms(realmNames).request(),
@@ -340,7 +346,7 @@ private <Result> void refreshRealms(ActionListener<Result> listener, Result resu
                             listener.onResponse(result);
                         },
                         ex -> {
-                            logger.warn("Failed to clear cache for realms [{}]", Arrays.toString(realmNames));
+                            logger.warn(new ParameterizedMessage("Failed to clear cache for realms [{}]", Arrays.toString(realmNames)), ex);
                             listener.onFailure(ex);
                         }),
                 securityClient::clearRealmCache);

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/support/mapper/NativeRoleMappingStoreTests.java

@@ -143,7 +143,7 @@ private SecurityIndexManager.State dummyState(ClusterHealthStatus indexStatus) {
 
     public void testCacheClearOnIndexHealthChange() {
         final AtomicInteger numInvalidation = new AtomicInteger(0);
-        final NativeRoleMappingStore store = buildRoleMappingStoreForInvalidationTesting(numInvalidation);
+        final NativeRoleMappingStore store = buildRoleMappingStoreForInvalidationTesting(numInvalidation, true);
 
         int expectedInvalidation = 0;
         // existing to no longer present
@@ -180,7 +180,7 @@ public void testCacheClearOnIndexHealthChange() {
 
     public void testCacheClearOnIndexOutOfDateChange() {
         final AtomicInteger numInvalidation = new AtomicInteger(0);
-        final NativeRoleMappingStore store = buildRoleMappingStoreForInvalidationTesting(numInvalidation);
+        final NativeRoleMappingStore store = buildRoleMappingStoreForInvalidationTesting(numInvalidation, true);
 
         store.onSecurityIndexStateChange(
             new SecurityIndexManager.State(Instant.now(), false, true, true, null, concreteSecurityIndexName, null),
@@ -193,40 +193,59 @@ public void testCacheClearOnIndexOutOfDateChange() {
         assertEquals(2, numInvalidation.get());
     }
 
-    private NativeRoleMappingStore buildRoleMappingStoreForInvalidationTesting(AtomicInteger invalidationCounter) {
+    public void testCacheIsNotClearedIfNoRealmsAreAttached() {
+        final AtomicInteger numInvalidation = new AtomicInteger(0);
+        final NativeRoleMappingStore store = buildRoleMappingStoreForInvalidationTesting(numInvalidation, false);
+
+        final SecurityIndexManager.State noIndexState = dummyState(null);
+        final SecurityIndexManager.State greenIndexState = dummyState(ClusterHealthStatus.GREEN);
+        store.onSecurityIndexStateChange(noIndexState, greenIndexState);
+        assertEquals(0, numInvalidation.get());
+    }
+
+    private NativeRoleMappingStore buildRoleMappingStoreForInvalidationTesting(AtomicInteger invalidationCounter, boolean attachRealm) {
         final Settings settings = Settings.builder().put("path.home", createTempDir()).build();
 
         final ThreadPool threadPool = mock(ThreadPool.class);
         final ThreadContext threadContext = new ThreadContext(settings);
         when(threadPool.getThreadContext()).thenReturn(threadContext);
 
+        final String realmName = randomAlphaOfLengthBetween(4, 8);
+
         final Client client = mock(Client.class);
         when(client.threadPool()).thenReturn(threadPool);
         when(client.settings()).thenReturn(settings);
         doAnswer(invocationOnMock -> {
+            assertThat(invocationOnMock.getArguments(), Matchers.arrayWithSize(3));
+            final ClearRealmCacheRequest request = (ClearRealmCacheRequest) invocationOnMock.getArguments()[1];
+            assertThat(request.realms(), Matchers.arrayContaining(realmName));
+
             ActionListener<ClearRealmCacheResponse> listener = (ActionListener<ClearRealmCacheResponse>) invocationOnMock.getArguments()[2];
             invalidationCounter.incrementAndGet();
             listener.onResponse(new ClearRealmCacheResponse(new ClusterName("cluster"), Collections.emptyList(), Collections.emptyList()));
             return null;
         }).when(client).execute(eq(ClearRealmCacheAction.INSTANCE), any(ClearRealmCacheRequest.class), any(ActionListener.class));
 
-        final Environment env = TestEnvironment.newEnvironment(settings);
-        final RealmConfig realmConfig = new RealmConfig(new RealmConfig.RealmIdentifier("ldap", getTestName()),
-                settings, env, threadContext);
-        final CachingUsernamePasswordRealm mockRealm = new CachingUsernamePasswordRealm(realmConfig, threadPool) {
-            @Override
-            protected void doAuthenticate(UsernamePasswordToken token, ActionListener<AuthenticationResult> listener) {
-                listener.onResponse(AuthenticationResult.notHandled());
-            }
-
-            @Override
-            protected void doLookupUser(String username, ActionListener<User> listener) {
-                listener.onResponse(null);
-            }
-        };
         final NativeRoleMappingStore store = new NativeRoleMappingStore(Settings.EMPTY, client, mock(SecurityIndexManager.class),
             mock(ScriptService.class));
-        store.refreshRealmOnChange(mockRealm);
+
+        if (attachRealm) {
+            final Environment env = TestEnvironment.newEnvironment(settings);
+            final RealmConfig.RealmIdentifier identifier = new RealmConfig.RealmIdentifier("ldap", realmName);
+            final RealmConfig realmConfig = new RealmConfig(identifier, settings, env, threadContext);
+            final CachingUsernamePasswordRealm mockRealm = new CachingUsernamePasswordRealm(realmConfig, threadPool) {
+                @Override
+                protected void doAuthenticate(UsernamePasswordToken token, ActionListener<AuthenticationResult> listener) {
+                    listener.onResponse(AuthenticationResult.notHandled());
+                }
+
+                @Override
+                protected void doLookupUser(String username, ActionListener<User> listener) {
+                    listener.onResponse(null);
+                }
+            };
+            store.refreshRealmOnChange(mockRealm);
+        }
         return store;
     }
 }