From a6109063a21220f5c75b6d79ff66cb2694407863 Mon Sep 17 00:00:00 2001
From: Yang Wang <yang.wang@elastic.co>
Date: Wed, 8 Jul 2020 16:43:07 +1000
Subject: [PATCH] Even more robust test for API key auth 429 response (#59159)
 (#59208)

Ensure blocking tasks are running before submitting more no-op tasks. This ensures no task would be popped out of the queue unexpectedly, which in turn guarantees the rejection of subsequent authentication request.
---
 .../xpack/security/authc/ApiKeyIntegTests.java      | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/ApiKeyIntegTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/ApiKeyIntegTests.java
index 502586d0d89..0652da94ebd 100644
--- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/ApiKeyIntegTests.java
+++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/ApiKeyIntegTests.java
@@ -858,10 +858,10 @@ public class ApiKeyIntegTests extends SecurityIntegTestCase {
         assertApiKeyNotCreated(client,"key-5");
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/59149")
     public void testAuthenticationReturns429WhenThreadPoolIsSaturated() throws IOException, InterruptedException, ExecutionException {
         final String nodeName = randomFrom(internalCluster().getNodeNames());
         final Settings settings = internalCluster().getInstance(Settings.class, nodeName);
+        final int allocatedProcessors = EsExecutors.allocatedProcessors(settings);
         final ThreadPool threadPool = internalCluster().getInstance(ThreadPool.class, nodeName);
 
         final RoleDescriptor descriptor = new RoleDescriptor("auth_only", new String[] { }, null, null);
@@ -880,9 +880,9 @@ public class ApiKeyIntegTests extends SecurityIntegTestCase {
         final List<NodeInfo> nodeInfos = client().admin().cluster().prepareNodesInfo().get().getNodes().stream()
             .filter(nodeInfo -> nodeInfo.getNode().getName().equals(nodeName))
             .collect(Collectors.toList());
+        assertEquals(1, nodeInfos.size());
 
         final ExecutorService executorService = threadPool.executor(SECURITY_CRYPTO_THREAD_POOL_NAME);
-        final int allocatedProcessors = EsExecutors.allocatedProcessors(settings);
         final int numberOfThreads = (allocatedProcessors + 1) / 2;
         final CountDownLatch blockingLatch = new CountDownLatch(1);
         final CountDownLatch readyLatch = new CountDownLatch(numberOfThreads);
@@ -896,15 +896,18 @@ public class ApiKeyIntegTests extends SecurityIntegTestCase {
                 }
             });
         }
-        // Fill the whole queue for the crypto thread pool
+        // Make sure above tasks are running
+        readyLatch.await();
+        // Then fill the whole queue for the crypto thread pool
         Future<?> lastTaskFuture = null;
+        int i = 0;
         try {
-            for (int i = 0; i < CRYPTO_THREAD_POOL_QUEUE_SIZE; i++) {
+            for (i = 0; i < CRYPTO_THREAD_POOL_QUEUE_SIZE; i++) {
                 lastTaskFuture = executorService.submit(() -> { });
             }
         } catch (EsRejectedExecutionException e) {
+            logger.info("Attempted to push {} tasks but only pushed {}", CRYPTO_THREAD_POOL_QUEUE_SIZE, i + 1);
         }
-        readyLatch.await();
 
         try (RestClient restClient = createRestClient(nodeInfos, null, "http")) {
             final String base64ApiKeyKeyValue = Base64.getEncoder().encodeToString(