actions, String customStatus) {
this.actions = Collections.unmodifiableCollection(actions);
;
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/runner/ActivityRunner.java b/durabletask-client/src/main/java/io/dapr/durabletask/runner/ActivityRunner.java
new file mode 100644
index 0000000000..c9c0baa434
--- /dev/null
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/runner/ActivityRunner.java
@@ -0,0 +1,182 @@
+/*
+ * Copyright 2026 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.durabletask.runner;
+
+import com.google.protobuf.StringValue;
+import io.dapr.durabletask.FailureDetails;
+import io.dapr.durabletask.TaskActivityExecutor;
+import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
+import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
+import io.grpc.StatusRuntimeException;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.context.propagation.TextMapGetter;
+
+import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class ActivityRunner extends DurableRunner {
+ private static final Logger logger = Logger.getLogger(ActivityRunner.class.getPackage().getName());
+
+ private final OrchestratorService.ActivityRequest activityRequest;
+ private final TaskActivityExecutor taskActivityExecutor;
+
+ /**
+ * Constructor.
+ *
+ * This class executes the activity requests
+ *
+ * @param workItem work item to be executed
+ * @param taskActivityExecutor executor for the activity
+ * @param sidecarClient sidecar client to communicate with the sidecar
+ * @param tracer tracer to be used for tracing
+ */
+ public ActivityRunner(
+ OrchestratorService.WorkItem workItem,
+ TaskActivityExecutor taskActivityExecutor,
+ TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient,
+ @Nullable Tracer tracer) {
+ super(workItem, sidecarClient, tracer);
+ this.activityRequest = workItem.getActivityRequest();
+ this.taskActivityExecutor = taskActivityExecutor;
+ }
+
+ @Override
+ public void run() {
+ if (tracer != null) {
+ runWithTracing();
+ } else {
+ runWithoutTracing();
+ }
+ }
+
+ private void runWithTracing() {
+ Context parentContext = extractTraceContext();
+
+ Span span = tracer.spanBuilder("activity:" + activityRequest.getName())
+ .setParent(parentContext)
+ .setSpanKind(SpanKind.INTERNAL)
+ .setAttribute("durabletask.task.instance_id",
+ activityRequest.getOrchestrationInstance().getInstanceId())
+ .setAttribute("durabletask.task.id", activityRequest.getTaskId())
+ .setAttribute("durabletask.activity.name", activityRequest.getName())
+ .startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ executeActivity();
+ } catch (Throwable e) {
+ logger.log(Level.WARNING, "Failed to complete activity task.", e);
+ span.setStatus(StatusCode.ERROR, "Failed to complete activity task");
+ span.recordException(e);
+ } finally {
+ span.end();
+ }
+ }
+
+ private void runWithoutTracing() {
+ try {
+ executeActivity();
+ } catch (Throwable e) {
+ logger.log(Level.WARNING, "Failed to complete activity task.", e);
+ }
+ }
+
+ private void executeActivity() throws Throwable {
+ String output = null;
+ OrchestratorService.TaskFailureDetails failureDetails = null;
+ Throwable failureException = null;
+ try {
+ output = taskActivityExecutor.execute(
+ activityRequest.getName(),
+ activityRequest.getInput().getValue(),
+ activityRequest.getTaskExecutionId(),
+ activityRequest.getTaskId(),
+ activityRequest.getParentTraceContext().getTraceParent());
+ } catch (Throwable e) {
+ failureDetails = OrchestratorService.TaskFailureDetails.newBuilder()
+ .setErrorType(e.getClass().getName())
+ .setErrorMessage(e.getMessage())
+ .setStackTrace(StringValue.of(FailureDetails.getFullStackTrace(e)))
+ .build();
+ failureException = e;
+ }
+
+ OrchestratorService.ActivityResponse.Builder responseBuilder = OrchestratorService.ActivityResponse
+ .newBuilder()
+ .setInstanceId(activityRequest.getOrchestrationInstance().getInstanceId())
+ .setTaskId(activityRequest.getTaskId())
+ .setCompletionToken(workItem.getCompletionToken());
+
+ if (output != null) {
+ responseBuilder.setResult(StringValue.of(output));
+ }
+
+ if (failureDetails != null) {
+ responseBuilder.setFailureDetails(failureDetails);
+ }
+
+ try {
+ this.sidecarClient.completeActivityTask(responseBuilder.build());
+ } catch (StatusRuntimeException e) {
+ logException(e);
+ throw e;
+ }
+
+ if (failureException != null) {
+ throw failureException;
+ }
+ }
+
+ private Context extractTraceContext() {
+ if (!activityRequest.hasParentTraceContext()) {
+ return Context.current();
+ }
+
+ OrchestratorService.TraceContext traceContext = activityRequest.getParentTraceContext();
+ String traceParent = traceContext.getTraceParent();
+
+ if (traceParent.isEmpty()) {
+ return Context.current();
+ }
+
+ Map carrier = new HashMap<>();
+ carrier.put("traceparent", traceParent);
+ if (traceContext.hasTraceState()) {
+ carrier.put("tracestate", traceContext.getTraceState().getValue());
+ }
+
+ TextMapGetter> getter = new TextMapGetter<>() {
+ @Override
+ public Iterable keys(Map carrier) {
+ return carrier.keySet();
+ }
+
+ @Override
+ public String get(Map carrier, String key) {
+ return carrier.get(key);
+ }
+ };
+
+ return W3CTraceContextPropagator.getInstance()
+ .extract(Context.current(), carrier, getter);
+ }
+}
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/runner/DurableRunner.java b/durabletask-client/src/main/java/io/dapr/durabletask/runner/DurableRunner.java
new file mode 100644
index 0000000000..b59aa58046
--- /dev/null
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/runner/DurableRunner.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2026 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.durabletask.runner;
+
+import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
+import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.opentelemetry.api.trace.Tracer;
+
+import javax.annotation.Nullable;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public abstract class DurableRunner implements Runnable {
+ private static final Logger logger = Logger.getLogger(DurableRunner.class.getPackage().getName());
+ public final TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient;
+ public final OrchestratorService.WorkItem workItem;
+ @Nullable
+ public final Tracer tracer;
+
+ /**
+ * Constructs a new instance of the DurableRunner.
+ *
+ * @param workItem the work item to be executed
+ * @param sidecarClient the sidecar client used to communicate with the durable task sidecar
+ * @param tracer the tracer used for tracing operations; can be null if tracing is not required
+ */
+ public DurableRunner(OrchestratorService.WorkItem workItem,
+ TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient,
+ @Nullable Tracer tracer) {
+ this.workItem = workItem;
+ this.sidecarClient = sidecarClient;
+ this.tracer = tracer;
+ }
+
+ protected String getSidecarAddress() {
+ return this.sidecarClient.getChannel().authority();
+ }
+
+ protected void logException(StatusRuntimeException e) {
+ if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) {
+ logger.log(Level.WARNING,
+ "The sidecar at address {0} is unavailable while completing the activity task.",
+ this.sidecarClient.getChannel().authority());
+ } else if (e.getStatus().getCode() == Status.Code.CANCELLED) {
+ logger.log(Level.WARNING,
+ "Durable Task worker has disconnected from {0} while completing the activity task.",
+ this.sidecarClient.getChannel().authority());
+ } else {
+ logger.log(Level.WARNING, "Unexpected failure completing the activity task at {0}.",
+ this.sidecarClient.getChannel().authority());
+ }
+ }
+}
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/runner/OrchestratorRunner.java b/durabletask-client/src/main/java/io/dapr/durabletask/runner/OrchestratorRunner.java
new file mode 100644
index 0000000000..56eb47f670
--- /dev/null
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/runner/OrchestratorRunner.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright 2026 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.durabletask.runner;
+
+import com.google.protobuf.StringValue;
+import io.dapr.durabletask.TaskOrchestrationExecutor;
+import io.dapr.durabletask.TaskOrchestratorResult;
+import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
+import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
+import io.grpc.StatusRuntimeException;
+import io.opentelemetry.api.trace.Tracer;
+
+import javax.annotation.Nullable;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class OrchestratorRunner extends DurableRunner {
+ private static final Logger logger = Logger.getLogger(OrchestratorRunner.class.getPackage().getName());
+
+ private final OrchestratorService.OrchestratorRequest orchestratorRequest;
+ private final TaskOrchestrationExecutor taskOrchestrationExecutor;
+
+ /**
+ * Constructs a new instance of the OrchestratorRunner class.
+ *
+ * @param workItem The work item containing details about the orchestrator task to be executed.
+ * @param taskOrchestrationExecutor The executor responsible for running task orchestration logic.
+ * @param sidecarClient The gRPC stub for communication with the Task Hub sidecar service.
+ * @param tracer An optional tracer used for distributed tracing, can be null.
+ */
+ public OrchestratorRunner(
+ OrchestratorService.WorkItem workItem,
+ TaskOrchestrationExecutor taskOrchestrationExecutor,
+ TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient,
+ @Nullable Tracer tracer) {
+
+ super(workItem, sidecarClient, tracer);
+ this.orchestratorRequest = workItem.getOrchestratorRequest();
+ this.taskOrchestrationExecutor = taskOrchestrationExecutor;
+ }
+
+ @Override
+ public void run() {
+ TaskOrchestratorResult taskOrchestratorResult = taskOrchestrationExecutor.execute(
+ orchestratorRequest.getPastEventsList(),
+ orchestratorRequest.getNewEventsList());
+
+ OrchestratorService.OrchestratorResponse response = OrchestratorService.OrchestratorResponse.newBuilder()
+ .setInstanceId(orchestratorRequest.getInstanceId())
+ .addAllActions(taskOrchestratorResult.getActions())
+ .setCustomStatus(StringValue.of(taskOrchestratorResult.getCustomStatus()))
+ .setCompletionToken(workItem.getCompletionToken())
+ .build();
+
+ try {
+ this.sidecarClient.completeOrchestratorTask(response);
+ logger.log(Level.FINEST,
+ "Completed orchestrator request for instance: {0}",
+ orchestratorRequest.getInstanceId());
+ } catch (StatusRuntimeException e) {
+ this.logException(e);
+ }
+ }
+}
diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowActivityContext.java b/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowActivityContext.java
index 2391b8f635..229caaf88c 100644
--- a/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowActivityContext.java
+++ b/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowActivityContext.java
@@ -25,4 +25,5 @@ public interface WorkflowActivityContext {
T getInput(Class targetType);
+ String getTraceParent();
}
diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowContext.java b/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowContext.java
index 8608e96937..be3280a992 100644
--- a/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowContext.java
+++ b/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowContext.java
@@ -20,7 +20,6 @@
import org.slf4j.Logger;
import javax.annotation.Nullable;
-
import java.time.Duration;
import java.time.Instant;
import java.time.ZonedDateTime;
@@ -383,7 +382,7 @@ default Task callChildWorkflow(String name) {
* @return a new {@link Task} that completes when the child-workflow completes or fails
*/
default Task callChildWorkflow(String name, Object input) {
- return this.callChildWorkflow(name, input, null);
+ return this.callChildWorkflow(name, input, Void.class);
}
/**
diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContext.java b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContext.java
index d6f935e8f8..e5d648ff57 100644
--- a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContext.java
+++ b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContext.java
@@ -97,6 +97,11 @@ public T getInput(Class targetType) {
return this.innerContext.getInput(targetType);
}
+ @Override
+ public String getTraceParent() {
+ return this.innerContext.getTraceParent();
+ }
+
@Override
public String getTaskExecutionId() {
return this.innerContext.getTaskExecutionId();
diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/DefaultWorkflowContextTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/DefaultWorkflowContextTest.java
index b6ca38ecbc..feb2025f3a 100644
--- a/sdk-workflows/src/test/java/io/dapr/workflows/DefaultWorkflowContextTest.java
+++ b/sdk-workflows/src/test/java/io/dapr/workflows/DefaultWorkflowContextTest.java
@@ -38,7 +38,11 @@
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class DefaultWorkflowContextTest {
private DefaultWorkflowContext context;
@@ -284,7 +288,7 @@ public void callChildWorkflowWithName() {
String expectedName = "TestActivity";
context.callChildWorkflow(expectedName);
- verify(mockInnerContext, times(1)).callSubOrchestrator(expectedName, null, null, null, null);
+ verify(mockInnerContext, times(1)).callSubOrchestrator(expectedName, null, null, null, Void.class);
}
@Test
diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContextTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContextTest.java
index 13cf3b6c6a..ceb6648e6c 100644
--- a/sdk-workflows/src/test/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContextTest.java
+++ b/sdk-workflows/src/test/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContextTest.java
@@ -5,7 +5,9 @@
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -21,6 +23,7 @@ void shouldSuccessfullyCreateContextAndReturnCorrectValuesForAllMethods() {
when(mockInnerContext.getName()).thenReturn("TestActivity");
when(mockInnerContext.getInput(any())).thenReturn("TestInput");
when(mockInnerContext.getTaskExecutionId()).thenReturn("TestExecutionId");
+ when(mockInnerContext.getTraceParent()).thenReturn("00244654132154564654");
assertNotNull(context.getLogger());
assertEquals("TestActivity", context.getName());
@@ -29,6 +32,7 @@ void shouldSuccessfullyCreateContextAndReturnCorrectValuesForAllMethods() {
assertEquals("TestInput", input);
assertEquals("TestExecutionId", context.getTaskExecutionId());
+ assertEquals("00244654132154564654", context.getTraceParent());
}
@Test