Repository: quarkusio/quarkus
Issue: Subsequent @QuarkusTest's stuck when running with different @TestProfile for testing Lambdas

**Describe the bug**
The Gradle project consists of multiple `RequestHandler` which all need to be tested individually. We use one test profile for each handler test and one test class per handler. 
The test profiles spawns test resources (WireMock, localstack) and sets the handler `quarkus.lambda.handler` accordingly with some other variables. 
When running the tests we observe that the first `@QuarkusTest` class runs as expected but the subsequent class get stuck on the `LambdaClient.invoke(...)` call and blocks infinitely.

After some investigation it seems like the running `AbstractLambdaPollLoop` of the first QuarkusTest does not receive/check the interrupt of the Quarkus shutdown due to TestProfile switch. 
Additionally the second QuarkusTest somehow does not wait until the old poll loop is destroyed and starts its poll loop and tests. The Lambda test call of the second QuarkusTest is queued into `LambdaClient.REQUEST_QUEUE`.
However, due to the fact that the first poll loop is still running, it requests another requestId in line `AbstractLambdaPollLoop.java:60` which is then dequeued of the REQUEST_QUEUE by `LambdaResourceManager`.
Since the loop has been stopped, it exits and does not propagate the response to the `CompletableFuture` inside `LambdaClient`. The test call however of the second QuarkusTest is still waiting for the response. 

**Expected behavior**
All tests run with their assigned profile without being blocking.

**Actual behavior**
A subsequent QuarkusTest blocks due to a not returning call inside `AbstractLambdaPollLoop`.


**To Reproduce**

We were not able to reproduce it in a small example project. However we hope that contributors of the lambda extension 
with more in-depth knowledge of the Quarkus testing lifecycle will understand the above. If not we can only provide an example
of the testing setup based on the aws-lambda archetype project which somehow does not run in the stated problem.

**Configuration**
```properties
quarkus.lambda.enable-polling-jvm-mode=true

```

**Screenshots**
(If applicable, add screenshots to help explain your problem.)

**Environment (please complete the following information):**
 - Output of `java -version`:
openjdk 13.0.2 2020-01-14
OpenJDK Runtime Environment (build 13.0.2+8)
OpenJDK 64-Bit Server VM (build 13.0.2+8, mixed mode, sharing)

 - Quarkus version or git rev: 
 
 1.11.3.Final

 - Build tool (ie. output of `mvnw --version` or `gradlew --version`): 
 
 Gradle 6.5.1

**Additional context**
(Add any other context about the problem here.)


Changed files:
extensions/amazon-lambda/common-runtime/src/main/java/io/quarkus/amazon/lambda/runtime/AbstractLambdaPollLoop.java
test-framework/amazon-lambda/src/main/java/io/quarkus/amazon/lambda/test/LambdaResourceManager.java

diff --git a/extensions/amazon-lambda/common-runtime/src/main/java/io/quarkus/amazon/lambda/runtime/AbstractLambdaPollLoop.java b/extensions/amazon-lambda/common-runtime/src/main/java/io/quarkus/amazon/lambda/runtime/AbstractLambdaPollLoop.java
index 720e8d4f452..0777ce720e8 100644
--- a/extensions/amazon-lambda/common-runtime/src/main/java/io/quarkus/amazon/lambda/runtime/AbstractLambdaPollLoop.java
+++ b/extensions/amazon-lambda/common-runtime/src/main/java/io/quarkus/amazon/lambda/runtime/AbstractLambdaPollLoop.java
@@ -36,7 +36,6 @@ public AbstractLambdaPollLoop(ObjectMapper objectMapper, ObjectReader cognitoIdR
     public void startPollLoop(ShutdownContext context) {
 
         final AtomicBoolean running = new AtomicBoolean(true);
-
         final Thread pollingThread = new Thread(new Runnable() {
             @SuppressWarnings("unchecked")
             @Override
@@ -122,8 +121,11 @@ public void run() {
                 }
             }
         }, "Lambda Thread");
+        pollingThread.setDaemon(true);
         context.addShutdownTask(() -> {
             running.set(false);
+            //note that this does not seem to be 100% reliable in unblocking the thread
+            //which is why it is a daemon.
             pollingThread.interrupt();
         });
         pollingThread.start();
diff --git a/test-framework/amazon-lambda/src/main/java/io/quarkus/amazon/lambda/test/LambdaResourceManager.java b/test-framework/amazon-lambda/src/main/java/io/quarkus/amazon/lambda/test/LambdaResourceManager.java
index 1ae03daefd9..73278d9e3bd 100644
--- a/test-framework/amazon-lambda/src/main/java/io/quarkus/amazon/lambda/test/LambdaResourceManager.java
+++ b/test-framework/amazon-lambda/src/main/java/io/quarkus/amazon/lambda/test/LambdaResourceManager.java
@@ -7,6 +7,7 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -14,6 +15,7 @@
 import io.quarkus.amazon.lambda.runtime.FunctionError;
 import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
 import io.undertow.Undertow;
+import io.undertow.httpcore.StatusCodes;
 import io.undertow.server.HttpHandler;
 import io.undertow.server.HttpServerExchange;
 import io.undertow.server.RoutingHandler;
@@ -21,27 +23,38 @@
 
 public class LambdaResourceManager implements QuarkusTestResourceLifecycleManager {
 
+    private volatile boolean started = false;
     private volatile Undertow undertow;
+    private final AtomicInteger currentPollCount = new AtomicInteger();
 
     public static final int PORT = Integer.getInteger("quarkus-internal.aws-lambda.test-port", 5387);
 
     @Override
     public Map<String, String> start() {
-
+        started = true;
         RoutingHandler routingHandler = new RoutingHandler(true);
         routingHandler.add("GET", AmazonLambdaApi.API_PATH_INVOCATION_NEXT, new HttpHandler() {
             @Override
             public void handleRequest(HttpServerExchange exchange) throws Exception {
-                LambdaStartedNotifier.started = true;
-                Map.Entry<String, String> req = null;
-                while (req == null) {
-                    req = LambdaClient.REQUEST_QUEUE.poll(100, TimeUnit.MILLISECONDS);
-                    if (undertow == null || undertow.getWorker() == null || undertow.getWorker().isShutdown()) {
-                        return;
+                currentPollCount.incrementAndGet();
+                try {
+                    LambdaStartedNotifier.started = true;
+                    Map.Entry<String, String> req = null;
+                    while (req == null) {
+                        req = LambdaClient.REQUEST_QUEUE.poll(100, TimeUnit.MILLISECONDS);
+                        if (!started || undertow == null || undertow.getWorker() == null || undertow.getWorker().isShutdown()) {
+                            exchange.setStatusCode(StatusCodes.SERVICE_UNAVAILABLE);
+                            exchange.setPersistent(false);
+                            exchange.getOutputStream().close();
+                            return;
+                        }
                     }
+                    exchange.addResponseHeader(AmazonLambdaApi.LAMBDA_RUNTIME_AWS_REQUEST_ID, req.getKey());
+                    exchange.getOutputStream().write(req.getValue().getBytes(StandardCharsets.UTF_8));
+                    exchange.getOutputStream().close();
+                } finally {
+                    currentPollCount.decrementAndGet();
                 }
-                exchange.addResponseHeader(AmazonLambdaApi.LAMBDA_RUNTIME_AWS_REQUEST_ID, req.getKey());
-                exchange.writeAsync(req.getValue());
             }
         });
         routingHandler.add("POST", AmazonLambdaApi.API_PATH_INVOCATION + "{req}" + AmazonLambdaApi.API_PATH_RESPONSE,
@@ -117,6 +130,16 @@ public void handleRequest(HttpServerExchange exchange) throws Exception {
 
     @Override
     public void stop() {
+        started = false;
+        while (currentPollCount.get() > 0) {
+            try {
+                //wait for all the responses to be sent
+                //before we shutdown
+                Thread.sleep(10);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }
         if (undertow == null)
             return;
         undertow.stop();