Move integration tests to DTS Emulator (#230)

Signed-off-by: Hal Spang <halspang@microsoft.com>

Author: halspang

.github/workflows/build-validation.yml

@@ -73,12 +73,11 @@ jobs:
       if: env.UNIT_TEST_FAILED == 'true'
       run: exit 1
 
-      # TODO: Move the sidecar into a central image repository
-    - name: Initialize Durable Task Sidecar
-      run: docker run --name durabletask-sidecar -p 4001:4001 --env 'DURABLETASK_SIDECAR_LOGLEVEL=Debug' -d peterstone2019/durabletask-sidecar:latest start --backend Emulator
+    - name: Initialize Durable Task Emulator
+      run: docker run --name durabletask-emulator -p 4001:8080 -d mcr.microsoft.com/dts/dts-emulator:latest
 
-    - name: Display Durable Task Sidecar Logs
-      run: nohup docker logs --since=0 durabletask-sidecar > durabletask-sidecar.log 2>&1 &
+    - name: Display Durable Task Emulator Logs
+      run: nohup docker logs --since=0 durabletask-emulator > durabletask-emulator.log 2>&1 &
 
       # wait for 10 seconds, so sidecar container can be fully up, this will avoid intermittent failing issues for integration tests causing by failed to connect to sidecar
     - name: Wait for 10 seconds
@@ -88,14 +87,14 @@ jobs:
       run: ./gradlew integrationTest || echo "TEST_FAILED=true" >> $GITHUB_ENV
       continue-on-error: true
 
-    - name: Kill Durable Task Sidecar
-      run: docker kill durabletask-sidecar
+    - name: Kill Durable Task Emulator
+      run: docker kill durabletask-emulator
 
-    - name: Upload Durable Task Sidecar Logs
+    - name: Upload Durable Task Emulator Logs
       uses: actions/upload-artifact@v4
       with:
-        name: Durable Task Sidecar Logs
-        path: durabletask-sidecar.log
+        name: Durable Task Emulator Logs
+        path: durabletask-emulator.log
 
     - name: Archive test report
       uses: actions/upload-artifact@v4

azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientExtensions.java

@@ -91,7 +91,8 @@ public static DurableTaskGrpcClientBuilder createClientBuilder(
         return createBuilderFromOptions(new DurableTaskSchedulerClientOptions()
             .setEndpointAddress(endpoint)
             .setTaskHubName(taskHubName)
-            .setCredential(tokenCredential));
+            .setCredential(tokenCredential)
+            .setAllowInsecureCredentials(tokenCredential == null));
     }
 
     // Private helper methods to reduce code duplication

azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerExtensions.java

@@ -91,7 +91,8 @@ public static DurableTaskGrpcWorkerBuilder createWorkerBuilder(
         return createBuilderFromOptions(new DurableTaskSchedulerWorkerOptions()
             .setEndpointAddress(endpoint)
             .setTaskHubName(taskHubName)
-            .setCredential(tokenCredential));
+            .setCredential(tokenCredential)
+            .setAllowInsecureCredentials(tokenCredential == null));
     }
 
     // Private helper methods to reduce code duplication

client/build.gradle

@@ -17,6 +17,8 @@ def grpcVersion = '1.59.0'
 def protocVersion = '3.12.0'
 def jacksonVersion = '2.15.3'
 def openTelemetryVersion = '1.25.0'
+def azureCoreVersion = '1.45.0'
+def azureIdentityVersion = '1.11.1'
 // When build on local, you need to set this value to your local jdk11 directory.
 // Java11 is used to compile and run all the tests.
 // Example for Windows:  C:/Program Files/Java/openjdk-11.0.12_7/
@@ -36,12 +38,14 @@ dependencies {
     implementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
     implementation "com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}"
     implementation "com.fasterxml.jackson.datatype:jackson-datatype-jsr310:${jacksonVersion}"
-
     implementation "io.opentelemetry:opentelemetry-api:${openTelemetryVersion}"
     implementation "io.opentelemetry:opentelemetry-context:${openTelemetryVersion}"
 
     testImplementation(platform('org.junit:junit-bom:5.7.2'))
     testImplementation('org.junit.jupiter:junit-jupiter')
+    testImplementation project(':azuremanaged')
+    testImplementation "com.azure:azure-core:${azureCoreVersion}"
+    testImplementation "com.azure:azure-identity:${azureIdentityVersion}"
 }
 
 compileJava {

client/src/test/java/com/microsoft/durabletask/ErrorHandlingIntegrationTests.java

@@ -13,7 +13,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.junit.jupiter.api.BeforeEach;
 import static org.junit.jupiter.api.Assertions.*;
 
 /**
@@ -26,13 +25,6 @@
 @Tag("integration")
 public class ErrorHandlingIntegrationTests extends IntegrationTestBase {
 
-    @BeforeEach
-    private void startUp() {
-        try(DurableTaskClient client = new DurableTaskGrpcClientBuilder().build()) {
-            client.deleteTaskHub();
-        }
-    }
-
     @Test
     void orchestratorException() throws TimeoutException {
         final String orchestratorName = "OrchestratorWithException";
@@ -44,7 +36,7 @@ void orchestratorException() throws TimeoutException {
                 })
                 .buildAndStart();
 
-        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+        DurableTaskClient client = this.createClientBuilder().build();
         try (worker; client) {
             String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, 0);
             OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
@@ -83,7 +75,7 @@ void activityException(boolean handleException) throws TimeoutException {
                 })
                 .buildAndStart();
 
-        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+        DurableTaskClient client = this.createClientBuilder().build();
         try (worker; client) {
             String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, "");
             OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
@@ -166,7 +158,7 @@ void subOrchestrationException(boolean handleException) throws TimeoutException
                     throw new RuntimeException(errorMessage);
                 })
                 .buildAndStart();
-        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+        DurableTaskClient client = this.createClientBuilder().build();
         try (worker; client) {
             String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, 1);
             OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
@@ -284,7 +276,7 @@ private FailureDetails retryOnFailuresCoreTest(
                 })
                 .buildAndStart();
 
-        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+        DurableTaskClient client = this.createClientBuilder().build();
         try (worker; client) {
             String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, "");
             OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);

client/src/test/java/com/microsoft/durabletask/IntegrationTestBase.java

@@ -7,21 +7,62 @@
 
 import java.time.Duration;
 
+import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerClientOptions;
+import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerWorkerExtensions;
+import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerWorkerOptions;
+
+import io.grpc.Channel;
+import io.grpc.ManagedChannel;
 public class IntegrationTestBase {
     protected static final Duration defaultTimeout = Duration.ofSeconds(10);
 
     // All tests that create a server should save it to this variable for proper shutdown
-    private DurableTaskGrpcWorker server;
+    private DurableTaskGrpcWorker server;    // All tests that create a client are responsible for closing their own gRPC channel
+    private ManagedChannel workerChannel;
+    private ManagedChannel clientChannel;
 
     @AfterEach
     private void shutdown() {
         if (this.server != null) {
             this.server.stop();
+            this.server = null;
+        }
+
+        if (this.workerChannel != null) {
+            this.workerChannel.shutdownNow();
+            this.workerChannel = null;
+        }
+
+        if (this.clientChannel != null) {
+            this.clientChannel.shutdownNow();
+            this.clientChannel = null;
         }
     }
 
     protected TestDurableTaskWorkerBuilder createWorkerBuilder() {
-        return new TestDurableTaskWorkerBuilder();
+        DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions()
+            .setEndpointAddress("http://localhost:4001")
+            .setTaskHubName("default")
+            .setCredential(null)
+            .setAllowInsecureCredentials(true);
+        Channel grpcChannel = options.createGrpcChannel();
+        this.workerChannel = (ManagedChannel) grpcChannel;
+        return new TestDurableTaskWorkerBuilder(
+            new DurableTaskGrpcWorkerBuilder()
+                .grpcChannel(grpcChannel));
+    }    
+    
+    protected DurableTaskGrpcClientBuilder createClientBuilder() {
+        DurableTaskSchedulerClientOptions options = new DurableTaskSchedulerClientOptions()
+            .setEndpointAddress("http://localhost:4001")
+            .setTaskHubName("default")
+            .setCredential(null)
+            .setAllowInsecureCredentials(true);
+        Channel grpcChannel = options.createGrpcChannel();
+        // The channel returned is actually a ManagedChannel, so we can safely cast it
+        this.clientChannel = (ManagedChannel) grpcChannel;
+        return new DurableTaskGrpcClientBuilder()
+            .grpcChannel(grpcChannel);
     }
 
     public class TestDurableTaskWorkerBuilder {
@@ -31,6 +72,10 @@ private TestDurableTaskWorkerBuilder() {
             this.innerBuilder = new DurableTaskGrpcWorkerBuilder();
         }
 
+        private TestDurableTaskWorkerBuilder(DurableTaskGrpcWorkerBuilder innerBuilder) {
+            this.innerBuilder = innerBuilder;
+        }
+
         public DurableTaskGrpcWorker buildAndStart() {
             DurableTaskGrpcWorker server = this.innerBuilder.build();
             IntegrationTestBase.this.server = server;