Fix Flaky Integration Tests + Migrate some tests to Test Containers

.github/workflows/build.yml

@@ -187,7 +187,7 @@ jobs:
         ./dist/linux_amd64/release/placement &
     - name: Spin local environment
       run: |
-        docker compose -f ./sdk-tests/deploy/local-test.yml up -d mongo kafka
+        docker compose -f ./sdk-tests/deploy/local-test.yml up -d kafka
         docker ps
     - name: Install local ToxiProxy to simulate connectivity issues to Dapr sidecar
       run: |

pom.xml

@@ -73,6 +73,7 @@
     <junit-platform-console.version>1.7.0</junit-platform-console.version>
     <reactor.version>3.5.12</reactor.version>
     <testcontainers-redis.version>2.2.4</testcontainers-redis.version>
+    <jedis.version>5.1.0</jedis.version>
     <slf4j.version>2.0.9</slf4j.version>
     <mockito.version>3.11.2</mockito.version>
     <opentelemetry-bom.version>2.1.0</opentelemetry-bom.version>
@@ -406,6 +407,12 @@
           </exclusion>
         </exclusions>
       </dependency>
+      <dependency>
+        <groupId>redis.clients</groupId>
+        <artifactId>jedis</artifactId>
+        <version>${jedis.version}</version>
+        <scope>test</scope>
+      </dependency>
       <dependency>
         <groupId>io.github.microcks</groupId>
         <artifactId>microcks-testcontainers</artifactId>

sdk-tests/components/secretstore.yaml

@@ -7,7 +7,7 @@ spec:
   version: v1
   metadata:
     - name: secretsFile
-      value: "./components/secret.json"
+      value: "./secrets/secret.json"
     - name: nestedSeparator
       value: ":"
     - name: multiValued

sdk-tests/deploy/local-test.yml

@@ -21,8 +21,3 @@ services:
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-
-  mongo:
-    image: mongo
-    ports:
-      - "27017:27017"

sdk-tests/pom.xml

@@ -152,6 +152,11 @@
       <groupId>org.springframework.data</groupId>
       <artifactId>spring-data-keyvalue</artifactId>
     </dependency>
+    <dependency>
+      <groupId>redis.clients</groupId>
+      <artifactId>jedis</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.wiremock</groupId>
       <artifactId>wiremock-standalone</artifactId>

sdk-tests/src/test/java/io/dapr/it/AppRun.java

@@ -50,6 +50,31 @@ public class AppRun implements Stoppable {
     this.maxWaitMilliseconds = maxWaitMilliseconds;
   }
 
+  /**
+   * Overload used by {@link io.dapr.it.containers.BaseContainerIT} when the Dapr
+   * sidecar runs in a Testcontainer rather than via {@code dapr run}. The
+   * {@code DAPR_HTTP_PORT} / {@code DAPR_GRPC_PORT} env vars on the spawned
+   * app process point at the explicit override values (typically the
+   * DaprContainer's mapped host ports) instead of {@code ports.getHttpPort() /
+   * ports.getGrpcPort()}.
+   */
+  public AppRun(DaprPorts ports,
+         String successMessage,
+         Class serviceClass,
+         int maxWaitMilliseconds,
+         int daprHttpPortOverride,
+         int daprGrpcPortOverride) {
+    this.command = new Command(
+            successMessage,
+            buildCommand(serviceClass, ports),
+            new HashMap<>() {{
+              put("DAPR_HTTP_PORT", Integer.toString(daprHttpPortOverride));
+              put("DAPR_GRPC_PORT", Integer.toString(daprGrpcPortOverride));
+            }});
+    this.ports = ports;
+    this.maxWaitMilliseconds = maxWaitMilliseconds;
+  }
+
   public void start() throws InterruptedException, IOException {
     long start = System.currentTimeMillis();
     // First, try to stop previous run (if left running).

sdk-tests/src/test/java/io/dapr/it/AppRunOverrideTest.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2025 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package io.dapr.it;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class AppRunOverrideTest {
+
+  /**
+   * Verifies that when we construct AppRun with explicit Dapr port overrides,
+   * the DAPR_HTTP_PORT / DAPR_GRPC_PORT env vars on the spawned command point
+   * at the override values, not at the DaprPorts-allocated ones.
+   */
+  @Test
+  void daprPortOverridesAreUsedInEnv() throws Exception {
+    DaprPorts ports = DaprPorts.build(true, true, true);
+    AppRun app = new AppRun(ports, "ready", Object.class, 1000, 12345, 67890);
+
+    Field commandField = AppRun.class.getDeclaredField("command");
+    commandField.setAccessible(true);
+    Command command = (Command) commandField.get(app);
+
+    Field envField = Command.class.getDeclaredField("env");
+    envField.setAccessible(true);
+    @SuppressWarnings("unchecked")
+    Map<String, String> env = (Map<String, String>) envField.get(command);
+
+    assertEquals("12345", env.get("DAPR_HTTP_PORT"));
+    assertEquals("67890", env.get("DAPR_GRPC_PORT"));
+  }
+}

sdk-tests/src/test/java/io/dapr/it/DaprRun.java

@@ -172,10 +172,13 @@ public void stop() throws InterruptedException, IOException {
     System.out.println("Stopping dapr application ...");
     try {
       this.stopCommand.run();
-
       System.out.println("Dapr application stopped.");
     } catch (RuntimeException e) {
-      System.out.println("Could not stop app " + this.appName + ": " + e.getMessage());
+      if (e.getMessage() != null && e.getMessage().contains("Could not find success criteria")) {
+        System.out.println("App " + this.appName + " already stopped or not found (ignored).");
+      } else {
+        System.out.println("Could not stop app " + this.appName + ": " + e.getMessage());
+      }
     }
   }
 
@@ -219,8 +222,7 @@ public void waitForAppHealth(int maxWaitMilliseconds) throws InterruptedExceptio
         while (System.currentTimeMillis() <= maxWait) {
           try {
             stub.healthCheck(Empty.getDefaultInstance());
-            // artursouza: workaround due to race condition with runtime's probe on app's health.
-            Thread.sleep(5000);
+            Thread.sleep(2000);
             return;
           } catch (Exception e) {
             Thread.sleep(1000);
@@ -232,29 +234,31 @@ public void waitForAppHealth(int maxWaitMilliseconds) throws InterruptedExceptio
         channel.shutdown();
       }
     } else {
-      Duration waitDuration = Duration.ofMillis(maxWaitMilliseconds);
+      long maxWait = System.currentTimeMillis() + maxWaitMilliseconds;
       HttpClient client = HttpClient.newBuilder()
           .version(HttpClient.Version.HTTP_1_1)
-          .connectTimeout(waitDuration)
+          .connectTimeout(Duration.ofSeconds(5))
           .build();
       String url = "http://127.0.0.1:" + this.getAppPort() + "/health";
       HttpRequest request = HttpRequest.newBuilder()
           .GET()
           .uri(URI.create(url))
           .build();
 
-      try {
-        HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
-
-        if (response.statusCode() != 200) {
-          throw new RuntimeException("error: HTTP service is not healthy.");
+      while (System.currentTimeMillis() <= maxWait) {
+        try {
+          HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
+          if (response.statusCode() == 200) {
+            Thread.sleep(2000);
+            return;
+          }
+        } catch (IOException e) {
+          // not ready yet
         }
-      } catch (IOException e) {
-          throw new RuntimeException("exception: HTTP service is not healthy.");
+        Thread.sleep(1000);
       }
 
-      // artursouza: workaround due to race condition with runtime's probe on app's health.
-      Thread.sleep(5000);
+      throw new RuntimeException("timeout: HTTP service is not healthy.");
     }
   }
 

sdk-tests/src/test/java/io/dapr/it/actors/ActivationDeactivationIT.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 The Dapr Authors
+ * Copyright 2025 The Dapr Authors
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -14,10 +14,14 @@
 package io.dapr.it.actors;
 
 import io.dapr.actors.ActorId;
+import io.dapr.actors.client.ActorClient;
 import io.dapr.actors.client.ActorProxyBuilder;
-import io.dapr.it.BaseIT;
+import io.dapr.it.AppRun;
 import io.dapr.it.actors.services.springboot.DemoActor;
 import io.dapr.it.actors.services.springboot.DemoActorService;
+import io.dapr.it.containers.BaseContainerIT;
+import io.dapr.testcontainers.DaprContainer;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -30,24 +34,38 @@
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
-public class ActivationDeactivationIT extends BaseIT {
+public class ActivationDeactivationIT extends BaseContainerIT {
 
   private static Logger logger = LoggerFactory.getLogger(ActivationDeactivationIT.class);
 
-  @Test
-  public void activateInvokeDeactivate() throws Exception {
-    // The call below will fail if service cannot start successfully.
-    var run = startDaprApp(
-        ActivationDeactivationIT.class.getSimpleName(),
-        DemoActorService.SUCCESS_MESSAGE,
+  private static DaprContainer dapr;
+  private static AppRun app;
+  private static ActorClient actorClient;
+
+  @BeforeAll
+  public static void start() throws Exception {
+    var pair = startAppAndAttach(
+        "activation-deactivation-it",
         DemoActorService.class,
-        true,
-        60000);
+        AppRun.AppProtocol.HTTP,
+        appPort -> {
+          DaprContainer d = daprBuilder("activation-deactivation-it")
+              .withAppPort(appPort)
+              .withAppChannelAddress("host.testcontainers.internal")
+              .withComponent(redisStateStore(STATE_STORE_NAME));
+          return d;
+        });
+    dapr = pair.dapr();
+    app = pair.app();
+    actorClient = newActorClient(dapr);
+    waitForActorsReady(dapr);
+  }
 
+  @Test
+  public void activateInvokeDeactivate() throws Exception {
     final AtomicInteger atomicInteger = new AtomicInteger(1);
     logger.debug("Creating proxy builder");
-    ActorProxyBuilder<DemoActor> proxyBuilder
-        = new ActorProxyBuilder(DemoActor.class, deferClose(run.newActorClient()));
+    ActorProxyBuilder<DemoActor> proxyBuilder = new ActorProxyBuilder(DemoActor.class, actorClient);
     logger.debug("Creating actorId");
     ActorId actorId1 = new ActorId(Integer.toString(atomicInteger.getAndIncrement()));
     logger.debug("Building proxy");
@@ -63,7 +81,7 @@ public void activateInvokeDeactivate() throws Exception {
     logger.debug("Retrieving active Actors");
     List<String> activeActors = proxy.retrieveActiveActors();
     logger.debug("Active actors: [" + activeActors.toString() + "]");
-    assertTrue(activeActors.contains(actorId1.toString()),"Expecting actorId:[" + actorId1.toString() + "]");
+    assertTrue(activeActors.contains(actorId1.toString()), "Expecting actorId:[" + actorId1.toString() + "]");
 
     ActorId actorId2 = new ActorId(Integer.toString(atomicInteger.getAndIncrement()));
     DemoActor proxy2 = proxyBuilder.build(actorId2);

sdk-tests/src/test/java/io/dapr/it/actors/ActorExceptionIT.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 The Dapr Authors
+ * Copyright 2025 The Dapr Authors
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -14,12 +14,13 @@
 package io.dapr.it.actors;
 
 import io.dapr.actors.ActorId;
+import io.dapr.actors.client.ActorClient;
 import io.dapr.actors.client.ActorProxyBuilder;
-import io.dapr.it.BaseIT;
-import io.dapr.it.DaprRun;
+import io.dapr.it.AppRun;
 import io.dapr.it.actors.app.MyActor;
 import io.dapr.it.actors.app.MyActorService;
-import org.junit.jupiter.api.Assertions;
+import io.dapr.it.containers.BaseContainerIT;
+import io.dapr.testcontainers.DaprContainer;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
@@ -30,44 +31,54 @@
 import static io.dapr.it.Retry.callWithRetry;
 import static io.dapr.it.TestUtils.assertThrowsDaprExceptionSubstring;
 
-
-public class ActorExceptionIT extends BaseIT {
+public class ActorExceptionIT extends BaseContainerIT {
 
   private static Logger logger = LoggerFactory.getLogger(ActorExceptionIT.class);
 
-  private static DaprRun run;
+  private static DaprContainer dapr;
+  private static AppRun app;
+  private static ActorClient actorClient;
 
   @BeforeAll
   public static void start() throws Exception {
-    // The call below will fail if service cannot start successfully.
-    run = startDaprApp(
-        ActorExceptionIT.class.getSimpleName(),
-        MyActorService.SUCCESS_MESSAGE,
+    var pair = startAppAndAttach(
+        "actor-exception-it",
         MyActorService.class,
-        true,
-        60000);
+        AppRun.AppProtocol.HTTP,
+        appPort -> {
+          DaprContainer d = daprBuilder("actor-exception-it")
+              .withAppPort(appPort)
+              .withAppChannelAddress("host.testcontainers.internal")
+              .withComponent(redisStateStore(STATE_STORE_NAME));
+          return d;
+        });
+    dapr = pair.dapr();
+    app = pair.app();
+    actorClient = newActorClient(dapr);
+    waitForActorsReady(dapr);
   }
 
   @Test
   public void exceptionTest() throws Exception {
     ActorProxyBuilder<MyActor> proxyBuilder =
-        new ActorProxyBuilder("MyActorTest", MyActor.class, deferClose(run.newActorClient()));
+        new ActorProxyBuilder("MyActorTest", MyActor.class, actorClient);
     MyActor proxy = proxyBuilder.build(new ActorId("1"));
 
     callWithRetry(() -> {
       assertThrowsDaprExceptionSubstring(
           "INTERNAL",
           "INTERNAL: error invoke actor method: error from actor service",
-          () ->  proxy.throwException());
+          () -> proxy.throwException());
     }, 10000);
   }
 
   @Test
   public void exceptionDueToMetadataTest() throws Exception {
     // Setting this HTTP header via actor metadata will cause the Actor HTTP server to error.
     Map<String, String> metadata = Map.of("Content-Length", "9999");
+    ActorClient metadataClient = newActorClient(dapr, metadata);
     ActorProxyBuilder<MyActor> proxyBuilderMetadataOverride =
-        new ActorProxyBuilder("MyActorTest", MyActor.class, deferClose(run.newActorClient(metadata)));
+        new ActorProxyBuilder("MyActorTest", MyActor.class, metadataClient);
 
     MyActor proxyWithMetadata = proxyBuilderMetadataOverride.build(new ActorId("2"));
     callWithRetry(() -> {