Context propagation - cleared Arc snapshot should destroy its state once the invocation ends

extensions/arc/runtime/src/main/java/io/quarkus/arc/runtime/context/ArcContextProvider.java

@@ -79,11 +79,11 @@ public ThreadContextController begin() {
                 // context active, store current state, start blank context anew and restore state afterwards
                 // it is not necessary to deactivate the context first - just overwrite the previous state
                 requestContext.activate();
-                return new RestoreContextController(requestContext, toRestore);
+                return new RestoreContextController(requestContext, toRestore, true);
             } else {
-                // context not active, activate blank one, deactivate afterwards
+                // context not active, activate blank one, terminate afterwards
                 requestContext.activate();
-                return requestContext::deactivate;
+                return requestContext::terminate;
             }
         }
     }
@@ -158,14 +158,24 @@ private static final class RestoreContextController implements ThreadContextCont
 
         private final ManagedContext requestContext;
         private final InjectableContext.ContextState stateToRestore;
+        private final boolean destroyRequestContext;
 
         RestoreContextController(ManagedContext requestContext, ContextState stateToRestore) {
+            this(requestContext, stateToRestore, false);
+        }
+
+        // in case of ClearContextSnapshot, we want to destroy instances of the intermediate context
+        RestoreContextController(ManagedContext requestContext, ContextState stateToRestore, boolean destroyRequestContext) {
             this.requestContext = requestContext;
             this.stateToRestore = stateToRestore;
+            this.destroyRequestContext = destroyRequestContext;
         }
 
         @Override
         public void endContext() throws IllegalStateException {
+            if (destroyRequestContext) {
+                requestContext.destroy();
+            }
             // it is not necessary to deactivate the context first - just overwrite the previous state
             requestContext.activate(stateToRestore.isValid() ? stateToRestore : null);
         }

integration-tests/smallrye-context-propagation/src/test/java/io/quarkus/context/test/RequestBean.java

@@ -1,11 +1,19 @@
 package io.quarkus.context.test;
 
+import javax.annotation.PreDestroy;
 import javax.enterprise.context.RequestScoped;
 
 @RequestScoped
 public class RequestBean {
 
+    public static volatile int DESTROY_INVOKED = 0;
+
     public String callMe() {
         return "Hello " + System.identityHashCode(this);
     }
+
+    @PreDestroy
+    public void destroy() {
+        DESTROY_INVOKED++;
+    }
 }

integration-tests/smallrye-context-propagation/src/test/java/io/quarkus/context/test/SimpleContextPropagationTest.java

@@ -8,6 +8,7 @@
 
 import org.awaitility.Awaitility;
 import org.awaitility.core.ThrowingRunnable;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
@@ -75,14 +76,20 @@ public void testArcTCContextPropagation() {
 
     @Test()
     public void testArcMEContextPropagationDisabled() {
+        // reset state
+        RequestBean.DESTROY_INVOKED = 0;
         RestAssured.when().get("/context/noarc").then()
                 .statusCode(Response.Status.OK.getStatusCode());
+        Assertions.assertEquals(2, RequestBean.DESTROY_INVOKED);
     }
 
     @Test()
     public void testArcTCContextPropagationDisabled() {
+        // reset state
+        RequestBean.DESTROY_INVOKED = 0;
         RestAssured.when().get("/context/noarc-tc").then()
                 .statusCode(Response.Status.OK.getStatusCode());
+        Assertions.assertEquals(2, RequestBean.DESTROY_INVOKED);
     }
 
     @Test()