test: fix ipc test

Author: jbutler

src/integrationtests/java/com/aws/greengrass/integrationtests/ipc/GetClientDeviceAuthTokenTest.java

@@ -5,6 +5,7 @@
 
 package com.aws.greengrass.integrationtests.ipc;
 
+import com.aws.greengrass.clientdevices.auth.infra.CDAExecutor;
 import com.aws.greengrass.dependency.State;
 import com.aws.greengrass.clientdevices.auth.ClientDevicesAuthService;
 import com.aws.greengrass.clientdevices.auth.exception.AuthenticationException;
@@ -27,6 +28,7 @@
 import org.junit.jupiter.api.extension.ExtensionContext;
 import org.junit.jupiter.api.io.TempDir;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.junit.jupiter.MockitoExtension;
 import software.amazon.awssdk.aws.greengrass.GetClientDeviceAuthTokenResponseHandler;
 import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClient;
@@ -47,21 +49,19 @@
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
-import static com.aws.greengrass.componentmanager.KernelConfigResolver.CONFIGURATION_CONFIG_KEY;
-import static com.aws.greengrass.clientdevices.auth.ClientDevicesAuthService.CLOUD_REQUEST_QUEUE_SIZE_TOPIC;
-import static com.aws.greengrass.clientdevices.auth.ClientDevicesAuthService.PERFORMANCE_TOPIC;
 import static com.aws.greengrass.testcommons.testutilities.ExceptionLogProtector.ignoreExceptionOfType;
 import static com.aws.greengrass.testcommons.testutilities.TestUtils.asyncAssertOnConsumer;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.ArgumentMatchers.anyMap;
-import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.when;
 
 @ExtendWith({GGExtension.class, UniqueRootPathExtension.class, MockitoExtension.class})
@@ -159,8 +159,15 @@ void GIVEN_brokerWithValidCredentials_WHEN_GetClientDeviceAuthToken_THEN_returns
     @Test
     void GIVEN_brokerWithInvalidCredentials_WHEN_GetClientDeviceAuthToken_THEN_throwsInvalidArgumentsError_and_WHEN_queueIsFull_THEN_throwsServiceError()
             throws Exception {
+        // Inject work queue which will reject any work added
+        LinkedBlockingQueue<Runnable> mockQueue = Mockito.mock(LinkedBlockingQueue.class);
+        when(mockQueue.offer(any())).thenReturn(false);
+        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 1, TimeUnit.SECONDS, mockQueue);
+        kernel.getContext().put(CDAExecutor.class, new CDAExecutor(executor));
         kernel.getContext().put(SessionManager.class, sessionManager);
+
         startNucleusWithConfig("cda.yaml");
+
         try (EventStreamRPCConnection connection = IPCTestUtils.getEventStreamRpcConnection(kernel,
                 "BrokerWithGetClientDeviceAuthTokenPermission")) {
             GreengrassCoreIPCClient ipcClient = new GreengrassCoreIPCClient(connection);
@@ -174,11 +181,6 @@ void GIVEN_brokerWithInvalidCredentials_WHEN_GetClientDeviceAuthToken_THEN_throw
             assertThat(err.getCause().getMessage(), containsString("Invalid client device credentials"));
         }
 
-        // Update the cloud queue size to 1 so that we'll just reject the second request
-        kernel.findServiceTopic(ClientDevicesAuthService.CLIENT_DEVICES_AUTH_SERVICE_NAME)
-                .lookup(CONFIGURATION_CONFIG_KEY, PERFORMANCE_TOPIC, CLOUD_REQUEST_QUEUE_SIZE_TOPIC).withValue(1);
-        kernel.getContext().waitForPublishQueueToClear();
-
         // Verify that we get a good error that the request couldn't be queued
         try (EventStreamRPCConnection connection = IPCTestUtils.getEventStreamRpcConnection(kernel,
                 "BrokerWithGetClientDeviceAuthTokenPermission")) {
@@ -187,27 +189,10 @@ void GIVEN_brokerWithInvalidCredentials_WHEN_GetClientDeviceAuthToken_THEN_throw
                     new CredentialDocument().withMqttCredential(
                             new MQTTCredential().withClientId("some-client-id").withCertificatePem("VALID PEM")));
 
-            CountDownLatch cdl = new CountDownLatch(1);
-            when(sessionManager.createSession(anyString(), anyMap())).thenAnswer((a) -> {
-                cdl.countDown();
-                Thread.sleep(1_000); // slow down the first request so that the second will be rejected
-                return "uuid";
-            });
-            // Request 1 (immediately runs)
-            CompletableFuture<GetClientDeviceAuthTokenResponse> fut1 =
-                    ipcClient.getClientDeviceAuthToken(request, Optional.empty()).getResponse();
-            // Ensure the threadpool is actively blocked before we send the next requests to fill the queue and then
-            // overflow the queue.
-            cdl.await(2, TimeUnit.SECONDS);
-            // Request 2 (queued so that queue size is 1)
-            CompletableFuture<GetClientDeviceAuthTokenResponse> fut2 =
-                    ipcClient.getClientDeviceAuthToken(request, Optional.empty()).getResponse();
-            // Request 3 (expect rejection)
+            // Expect rejection
             Exception err = assertThrows(Exception.class, () -> clientDeviceAuthToken(ipcClient, request, (r) -> {}));
             assertThat(err.getCause().getMessage(), containsString("Unable to queue request"));
             assertEquals(ServiceError.class, err.getCause().getClass());
-            fut1.get(2, TimeUnit.SECONDS);
-            fut2.get(2, TimeUnit.SECONDS);
         }
     }
 

src/main/java/com/aws/greengrass/clientdevices/auth/ClientDevicesAuthService.java

@@ -156,10 +156,14 @@ private void updateInfrastructure(InfrastructureConfiguration infraConfig) {
     }
 
     private void initializeInfrastructure() {
-        BlockingQueue<Runnable> queue = new ResizableLinkedBlockingQueue<>(DEFAULT_WORK_QUEUE_DEPTH);
-        ThreadPoolExecutor executor = new ThreadPoolExecutor(DEFAULT_THREAD_POOL_SIZE,
-                DEFAULT_THREAD_POOL_SIZE, 60, TimeUnit.SECONDS, queue);
-        context.put(CDAExecutor.class, new CDAExecutor(executor));
+        // Don't re-inject this if it is already present
+        CDAExecutor cdaExecutor = context.getIfExists(CDAExecutor.class, null);
+        if (cdaExecutor == null) {
+            BlockingQueue<Runnable> queue = new ResizableLinkedBlockingQueue<>(DEFAULT_WORK_QUEUE_DEPTH);
+            ThreadPoolExecutor executor = new ThreadPoolExecutor(DEFAULT_THREAD_POOL_SIZE,
+                    DEFAULT_THREAD_POOL_SIZE, 60, TimeUnit.SECONDS, queue);
+            context.put(CDAExecutor.class, new CDAExecutor(executor));
+        }
     }
 
     private void initializeIPC() {