Fix testRefreshingMultipleTimesWithinWindowSucceeds (#39701)

Previously all the threads were writing the received tokens to a
HashSet. In cases with many threads, sometimes (1 every ~25 tests)
calling size() on the HashSet returned 2 even though it seemed to
contain only one String and there was no evidence from logging that
threadSecurityClient.refreshToken() ever returned a different
access or refresh token.

This commit changes the test to use a ConcurrentHashMap instead,
checking that we only received one pair of access token/refresh token
eventually. It also adds a check so that we won't take into consideration
tokens that are returned after 30s, hence not in the concurrent refresh
time window.

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

@@ -40,18 +40,18 @@ import org.elasticsearch.xpack.security.support.SecurityIndexManager;
 import org.junit.After;
 import org.junit.Before;
 
+import java.time.Clock;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.index.mapper.MapperService.SINGLE_MAPPING_NAME;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoTimeout;
@@ -391,12 +391,12 @@ public class TokenAuthIntegTests extends SecurityIntegTestCase {
     }
 
     public void testRefreshingMultipleTimesWithinWindowSucceeds() throws Exception {
+        final Clock clock = Clock.systemUTC();
         Client client = client().filterWithHeader(Collections.singletonMap("Authorization",
             UsernamePasswordToken.basicAuthHeaderValue(SecuritySettingsSource.TEST_USER_NAME,
                 SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING)));
         SecurityClient securityClient = new SecurityClient(client);
-        Set<String> refreshTokens = new HashSet<>();
+        final List<String> tokens = Collections.synchronizedList(new ArrayList<>());
-        Set<String> accessTokens = new HashSet<>();
         CreateTokenResponse createTokenResponse = securityClient.prepareCreateToken()
             .setGrantType("password")
             .setUsername(SecuritySettingsSource.TEST_USER_NAME)
@@ -409,6 +409,7 @@ public class TokenAuthIntegTests extends SecurityIntegTestCase {
         final CountDownLatch readyLatch = new CountDownLatch(numberOfThreads + 1);
         final CountDownLatch completedLatch = new CountDownLatch(numberOfThreads);
         AtomicBoolean failed = new AtomicBoolean();
+        final Instant t1 = clock.instant();
         for (int i = 0; i < numberOfThreads; i++) {
             threads.add(new Thread(() -> {
                 // Each thread gets its own client so that more than one nodes will be hit
@@ -427,8 +428,13 @@ public class TokenAuthIntegTests extends SecurityIntegTestCase {
                     return;
                 }
                 threadSecurityClient.refreshToken(refreshRequest, ActionListener.wrap(result -> {
-                    accessTokens.add(result.getTokenString());
+                    final Instant t2 = clock.instant();
-                    refreshTokens.add(result.getRefreshToken());
+                    if (t1.plusSeconds(30L).isBefore(t2)){
+                        logger.warn("Tokens [{}], [{}] were received more than 30 seconds after the request, not checking them",
+                            result.getTokenString(), result.getRefreshToken());
+                    } else {
+                        tokens.add(result.getTokenString() + result.getRefreshToken());
+                    }
                     logger.info("received access token [{}] and refresh token [{}]", result.getTokenString(), result.getRefreshToken());
                     completedLatch.countDown();
                 }, e -> {
@@ -448,8 +454,8 @@ public class TokenAuthIntegTests extends SecurityIntegTestCase {
         }
         completedLatch.await();
         assertThat(failed.get(), equalTo(false));
-        assertThat(accessTokens.size(), equalTo(1));
+        // Assert that we only ever got one access_token/refresh_token pair
-        assertThat(refreshTokens.size(), equalTo(1));
+        assertThat(tokens.stream().distinct().collect(Collectors.toList()).size(), equalTo(1));
     }
 
     public void testRefreshAsDifferentUser() {