wip cross app call activity for wf

Signed-off-by: Cassandra Coyle <[email protected]>

Author: cicoyle

CONTRIBUTING.md

@@ -19,4 +19,19 @@ When updating the protobuf definitions in `internal/durabletask-protobuf/protos/
 
 ```shell
 ./gradlew generateProto
-```
+```
+
+## Test locally from dapr/java-sdk
+
+```shell
+./gradlew publishToMavenLocal
+```
+
+or simply `./gradlew build publishToMavenLocal`
+
+Check if it was released locally with:
+```shell
+ls ~/.m2/repository/io/dapr/durabletask-client/ 
+```
+
+Then update the durabletask-client in the java-sdk pom.xml file to the newest version you released locally.

client/build.gradle

@@ -26,6 +26,7 @@ dependencies {
     // https://github.com/grpc/grpc-java#download
     implementation "io.grpc:grpc-protobuf:${grpcVersion}"
     implementation "io.grpc:grpc-stub:${grpcVersion}"
+    implementation 'com.google.protobuf:protobuf-java:3.25.5'
     runtimeOnly "io.grpc:grpc-netty-shaded:${grpcVersion}"
 
     compileOnly "org.apache.tomcat:annotations-api:6.0.53"
@@ -194,6 +195,10 @@ java {
     withJavadocJar()
 }
 
+tasks.named('sourcesJar').configure {
+    dependsOn tasks.named('generateProto')
+}
+
 spotbugs {
     toolVersion = '4.9.2'
     effort = 'max'

client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java

@@ -36,6 +36,7 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
     private final DataConverter dataConverter;
     private final Duration maximumTimerInterval;
     private final ExecutorService workerPool;
+    private final String appId; // App ID for cross-app routing
 
     private final TaskHubSidecarServiceBlockingStub sidecarClient;
     private final boolean isExecutorServiceManaged;
@@ -45,6 +46,7 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
     DurableTaskGrpcWorker(DurableTaskGrpcWorkerBuilder builder) {
         this.orchestrationFactories.putAll(builder.orchestrationFactories);
         this.activityFactories.putAll(builder.activityFactories);
+        this.appId = builder.appId;
 
         Channel sidecarGrpcChannel;
         if (builder.channel != null) {
@@ -128,7 +130,8 @@ public void startAndBlock() {
                 this.orchestrationFactories,
                 this.dataConverter,
                 this.maximumTimerInterval,
-                logger);
+                logger,
+                this.appId);
         TaskActivityExecutor taskActivityExecutor = new TaskActivityExecutor(
                 this.activityFactories,
                 this.dataConverter,
@@ -143,6 +146,9 @@ public void startAndBlock() {
                     RequestCase requestType = workItem.getRequestCase();
                     if (requestType == RequestCase.ORCHESTRATORREQUEST) {
                         OrchestratorRequest orchestratorRequest = workItem.getOrchestratorRequest();
+                        logger.log(Level.FINEST,
+                                String.format("Processing orchestrator request for instance: {0}",
+                            orchestratorRequest.getInstanceId()));
 
                         // TODO: Error handling
                         this.workerPool.submit(() -> {
@@ -159,6 +165,9 @@ public void startAndBlock() {
 
                             try {
                                 this.sidecarClient.completeOrchestratorTask(response);
+                                logger.log(Level.FINEST,
+                                        "Completed orchestrator request for instance: {0}",
+                                    orchestratorRequest.getInstanceId());
                             } catch (StatusRuntimeException e) {
                                 if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) {
                                     logger.log(Level.WARNING,
@@ -177,7 +186,12 @@ public void startAndBlock() {
                         });
                     } else if (requestType == RequestCase.ACTIVITYREQUEST) {
                         ActivityRequest activityRequest = workItem.getActivityRequest();
+                        logger.log(Level.FINEST,
+                                String.format("Processing activity request: %s for instance: %s}",
+                                    activityRequest.getName(),
+                                    activityRequest.getOrchestrationInstance().getInstanceId()));
 
+                        // TODO: Error handling
                         this.workerPool.submit(() -> {
                             String output = null;
                             TaskFailureDetails failureDetails = null;
@@ -227,7 +241,8 @@ public void startAndBlock() {
                     } else if (requestType == RequestCase.HEALTHPING) {
                         // No-op
                     } else {
-                        logger.log(Level.WARNING, "Received and dropped an unknown '{0}' work-item from the sidecar.",
+                        logger.log(Level.WARNING,
+                                "Received and dropped an unknown '{0}' work-item from the sidecar.",
                                 requestType);
                     }
                 }

client/src/main/java/io/dapr/durabletask/NewOrchestrationInstanceOptions.java

@@ -12,6 +12,7 @@ public final class NewOrchestrationInstanceOptions {
     private String instanceId;
     private Object input;
     private Instant startTime;
+    private String appID; // Target app ID for cross-app workflow routing
 
     /**
      * Default constructor for the {@link NewOrchestrationInstanceOptions} class.
@@ -71,6 +72,17 @@ public NewOrchestrationInstanceOptions setStartTime(Instant startTime) {
         return this;
     }
 
+    /**
+     * Sets the target app ID for cross-app workflow routing.
+     *
+     * @param appID the target app ID for cross-app routing
+     * @return this {@link NewOrchestrationInstanceOptions} object
+     */
+    public NewOrchestrationInstanceOptions setAppID(String appID) {
+        this.appID = appID;
+        return this;
+    }
+
     /**
      * Gets the user-specified version of the new orchestration.
      *
@@ -106,4 +118,22 @@ public Object getInput() {
     public Instant getStartTime() {
         return this.startTime;
     }
+
+    /**
+     * Gets the configured target app ID for cross-app workflow routing.
+     *
+     * @return the configured target app ID
+     */
+    public String getAppID() {
+        return this.appID;
+    }
+
+    /**
+     * Checks if an app ID is configured for cross-app routing.
+     *
+     * @return true if an app ID is configured, false otherwise
+     */
+    public boolean hasAppID() {
+        return this.appID != null && !this.appID.isEmpty();
+    }
 }

client/src/main/java/io/dapr/durabletask/TaskOrchestrationContext.java

@@ -36,6 +36,13 @@ public interface TaskOrchestrationContext {
      */
     String getInstanceId();
 
+    /**
+     * Gets the app ID of the current orchestration instance, if available.
+     * This is used for cross-app workflow routing.
+     * @return the app ID of the current orchestration instance, or null if not available
+     */
+    String getAppId();
+
     /**
      * Gets the current orchestration time in UTC.
      * @return the current orchestration time in UTC

internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH

@@ -1 +1 @@
-ce184193bcf2d4dcf643bcc934d3ea897c4c8c5c
+545663a4db0040bb02c642c27010e337b9be8d31

internal/durabletask-protobuf/protos/orchestrator_service.proto

@@ -14,7 +14,7 @@ import "google/protobuf/empty.proto";
 
 message TaskRouter {
     string source = 1; // orchestrationAppID
-    string target = 2; // appID
+    optional string target = 2; // appID
 }
 
 message OrchestrationInstance {
@@ -29,6 +29,7 @@ message ActivityRequest {
     OrchestrationInstance orchestrationInstance = 4;
     int32 taskId = 5;
     TraceContext parentTraceContext = 6;
+    string taskExecutionId = 7;
 }
 
 message ActivityResponse {
@@ -99,16 +100,19 @@ message TaskScheduledEvent {
     google.protobuf.StringValue version = 2;
     google.protobuf.StringValue input = 3;
     TraceContext parentTraceContext = 4;
+    string taskExecutionId = 5;
 }
 
 message TaskCompletedEvent {
     int32 taskScheduledId = 1;
     google.protobuf.StringValue result = 2;
+    string taskExecutionId = 3;
 }
 
 message TaskFailedEvent {
     int32 taskScheduledId = 1;
     TaskFailureDetails failureDetails = 2;
+    string taskExecutionId = 3;
 }
 
 message SubOrchestrationInstanceCreatedEvent {
@@ -263,6 +267,7 @@ message ScheduleTaskAction {
     google.protobuf.StringValue version = 2;
     google.protobuf.StringValue input = 3;
     optional TaskRouter router = 4;
+    string taskExecutionId = 5;
 }
 
 message CreateSubOrchestrationAction {