orchestrationFactories = new HashMap<>();
- orchestrationFactories.put("*", new TaskOrchestrationFactory() {
+ TaskOrchestrationFactories orchestrationFactories = new TaskOrchestrationFactories();
+ orchestrationFactories.addOrchestration(new TaskOrchestrationFactory() {
@Override
public String getName() {
return "*";
@@ -145,6 +146,16 @@ public String getName() {
public TaskOrchestration create() {
return orchestration;
}
+
+ @Override
+ public String getVersionName() {
+ return "";
+ }
+
+ @Override
+ public Boolean isLatestVersion() {
+ return false;
+ }
});
TaskOrchestrationExecutor taskOrchestrationExecutor = new TaskOrchestrationExecutor(
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/OrchestrationRuntimeStatus.java b/durabletask-client/src/main/java/io/dapr/durabletask/OrchestrationRuntimeStatus.java
index 1bdd33ab38..e9530ae815 100644
--- a/durabletask-client/src/main/java/io/dapr/durabletask/OrchestrationRuntimeStatus.java
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/OrchestrationRuntimeStatus.java
@@ -68,7 +68,12 @@ public enum OrchestrationRuntimeStatus {
/**
* The orchestration is in a suspended state.
*/
- SUSPENDED;
+ SUSPENDED,
+
+ /**
+ * The orchestration is in a stalled state.
+ */
+ STALLED;
static OrchestrationRuntimeStatus fromProtobuf(OrchestratorService.OrchestrationStatus status) {
switch (status) {
@@ -88,6 +93,8 @@ static OrchestrationRuntimeStatus fromProtobuf(OrchestratorService.Orchestration
return PENDING;
case ORCHESTRATION_STATUS_SUSPENDED:
return SUSPENDED;
+ case ORCHESTRATION_STATUS_STALLED:
+ return STALLED;
default:
throw new IllegalArgumentException(String.format("Unknown status value: %s", status));
}
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/Task.java b/durabletask-client/src/main/java/io/dapr/durabletask/Task.java
index a3f3313816..de2f13e871 100644
--- a/durabletask-client/src/main/java/io/dapr/durabletask/Task.java
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/Task.java
@@ -27,12 +27,14 @@
*
* Task{@literal <}int{@literal >} activityTask = ctx.callActivity("MyActivity", someInput, int.class);
*
+ *
* Orchestrator code uses the {@link #await()} method to block on the completion of the task and retrieve the result.
* If the task is not yet complete, the {@code await()} method will throw an {@link OrchestratorBlockedException}, which
* pauses the orchestrator's execution so that it can save its progress into durable storage and schedule any
* outstanding work. When the task is complete, the orchestrator will run again from the beginning and the next time
* the task's {@code await()} method is called, the result will be returned, or a {@link TaskFailedException} will be
* thrown if the result of the task was an unhandled exception.
+ *
* Note that orchestrator code must never catch {@code OrchestratorBlockedException} because doing so can cause the
* orchestration instance to get permanently stuck.
*
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/TaskActivityContext.java b/durabletask-client/src/main/java/io/dapr/durabletask/TaskActivityContext.java
index b2043b51ee..7a0d1ed1ee 100644
--- a/durabletask-client/src/main/java/io/dapr/durabletask/TaskActivityContext.java
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/TaskActivityContext.java
@@ -34,7 +34,6 @@ public interface TaskActivityContext {
*/
T getInput(Class targetType);
-
/**
* Gets the execution id of the current task activity.
*
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/TaskFailedException.java b/durabletask-client/src/main/java/io/dapr/durabletask/TaskFailedException.java
index 377eecb426..5362e830c7 100644
--- a/durabletask-client/src/main/java/io/dapr/durabletask/TaskFailedException.java
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/TaskFailedException.java
@@ -16,6 +16,7 @@
/**
* Exception that gets thrown when awaiting a {@link Task} for an activity or sub-orchestration that fails with an
* unhandled exception.
+ *
* Detailed information associated with a particular task failure can be retrieved
* using the {@link #getErrorDetails()} method.
*/
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationContext.java b/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationContext.java
index df0c95ec82..97d851b47f 100644
--- a/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationContext.java
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationContext.java
@@ -352,6 +352,15 @@ default void continueAsNew(Object input) {
*/
void continueAsNew(Object input, boolean preserveUnprocessedEvents);
+ /**
+ * Check if the given patch name can be applied to the orchestration.
+ *
+ * @param patchName The name of the patch to check.
+ * @return True if the given patch name can be applied to the orchestration, False otherwise.
+ */
+
+ boolean isPatched(String patchName);
+
/**
* Create a new Uuid that is safe for replay within an orchestration or operation.
*
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationExecutor.java b/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationExecutor.java
index 7a3436b036..ae239b15ca 100644
--- a/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationExecutor.java
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationExecutor.java
@@ -19,7 +19,11 @@
import io.dapr.durabletask.implementation.protobuf.OrchestratorService.ScheduleTaskAction.Builder;
import io.dapr.durabletask.interruption.ContinueAsNewInterruption;
import io.dapr.durabletask.interruption.OrchestratorBlockedException;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactories;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactory;
+import io.dapr.durabletask.orchestration.exception.VersionNotRegisteredException;
import io.dapr.durabletask.util.UuidGenerator;
+import org.apache.commons.lang3.StringUtils;
import javax.annotation.Nullable;
import java.time.Duration;
@@ -47,14 +51,14 @@
final class TaskOrchestrationExecutor {
private static final String EMPTY_STRING = "";
- private final HashMap orchestrationFactories;
+ private final TaskOrchestrationFactories orchestrationFactories;
private final DataConverter dataConverter;
private final Logger logger;
private final Duration maximumTimerInterval;
private final String appId;
public TaskOrchestrationExecutor(
- HashMap orchestrationFactories,
+ TaskOrchestrationFactories orchestrationFactories,
DataConverter dataConverter,
Duration maximumTimerInterval,
Logger logger,
@@ -79,6 +83,9 @@ public TaskOrchestratorResult execute(List pas
}
completed = true;
logger.finest("The orchestrator execution completed normally");
+ } catch (VersionNotRegisteredException versionNotRegisteredException) {
+ logger.warning("The orchestrator version is not registered: " + versionNotRegisteredException.toString());
+ context.setVersionNotRegistered();
} catch (OrchestratorBlockedException orchestratorBlockedException) {
logger.fine("The orchestrator has yielded and will await for new events.");
} catch (ContinueAsNewInterruption continueAsNewInterruption) {
@@ -87,7 +94,7 @@ public TaskOrchestratorResult execute(List pas
} catch (Exception e) {
// The orchestrator threw an unhandled exception - fail it
// TODO: What's the right way to log this?
- logger.warning("The orchestrator failed with an unhandled exception: " + e.toString());
+ logger.warning("The orchestrator failed with an unhandled exception: " + e);
context.fail(new FailureDetails(e));
}
@@ -97,12 +104,16 @@ public TaskOrchestratorResult execute(List pas
context.complete(null);
}
- return new TaskOrchestratorResult(context.pendingActions.values(), context.getCustomStatus());
+ return new TaskOrchestratorResult(context.pendingActions.values(),
+ context.getCustomStatus(),
+ context.version,
+ context.encounteredPatches);
}
private class ContextImplTask implements TaskOrchestrationContext {
private String orchestratorName;
+ private final List encounteredPatches = new ArrayList<>();
private String rawInput;
private String instanceId;
private Instant currentInstant;
@@ -127,6 +138,11 @@ private class ContextImplTask implements TaskOrchestrationContext {
private Object continuedAsNewInput;
private boolean preserveUnprocessedEvents;
private Object customStatus;
+ private final Map appliedPatches = new HashMap<>();
+ private final Map historyPatches = new HashMap<>();
+
+ private OrchestratorService.OrchestrationVersion orchestratorStartedVersion;
+ private String version;
public ContextImplTask(List pastEvents,
List newEvents) {
@@ -363,6 +379,34 @@ public Task callActivity(
return this.createAppropriateTask(taskFactory, options);
}
+ @Override
+ public boolean isPatched(String patchName) {
+ var isPatched = this.checkPatch(patchName);
+ if (isPatched) {
+ this.encounteredPatches.add(patchName);
+ }
+
+ return isPatched;
+ }
+
+ public boolean checkPatch(String patchName) {
+ if (this.appliedPatches.containsKey(patchName)) {
+ return this.appliedPatches.get(patchName);
+ }
+
+ if (this.historyPatches.containsKey(patchName)) {
+ this.appliedPatches.put(patchName, true);
+ return true;
+ }
+
+ if (this.isReplaying) {
+ this.appliedPatches.put(patchName, false);
+ return false;
+ }
+ this.appliedPatches.put(patchName, true);
+ return true;
+ }
+
@Override
public void continueAsNew(Object input, boolean preserveUnprocessedEvents) {
Helpers.throwIfOrchestratorComplete(this.isComplete);
@@ -438,7 +482,7 @@ public Task callSubOrchestrator(
if (input instanceof TaskOptions) {
throw new IllegalArgumentException("TaskOptions cannot be used as an input. "
- + "Did you call the wrong method overload?");
+ + "Did you call the wrong method overload?");
}
String serializedInput = this.dataConverter.serialize(input);
@@ -924,6 +968,7 @@ private void processEvent(OrchestratorService.HistoryEvent e) {
case ORCHESTRATORSTARTED:
Instant instant = DataConverter.getInstantFromTimestamp(e.getTimestamp());
this.setCurrentInstant(instant);
+ this.orchestratorStartedVersion = e.getOrchestratorStarted().getVersion();
this.logger.fine(() -> this.instanceId + ": Workflow orchestrator started");
break;
case ORCHESTRATORCOMPLETED:
@@ -938,18 +983,35 @@ private void processEvent(OrchestratorService.HistoryEvent e) {
this.logger.fine(() -> this.instanceId + ": Workflow execution started");
this.setAppId(e.getRouter().getSourceAppID());
+ if (this.orchestratorStartedVersion != null
+ && this.orchestratorStartedVersion.getPatchesCount() > 0) {
+ for (var patch : this.orchestratorStartedVersion.getPatchesList()) {
+ this.historyPatches.put(patch, true);
+ }
+ }
+
+ var versionName = "";
+ if (this.orchestratorStartedVersion != null
+ && !StringUtils.isEmpty(this.orchestratorStartedVersion.getName())) {
+ versionName = this.orchestratorStartedVersion.getName();
+ }
+
// Create and invoke the workflow orchestrator
TaskOrchestrationFactory factory = TaskOrchestrationExecutor.this.orchestrationFactories
- .get(executionStarted.getName());
+ .getOrchestrationFactory(executionStarted.getName(), versionName);
+
if (factory == null) {
// Try getting the default orchestrator
- factory = TaskOrchestrationExecutor.this.orchestrationFactories.get("*");
+ factory = TaskOrchestrationExecutor.this.orchestrationFactories
+ .getOrchestrationFactory("*");
}
// TODO: Throw if the factory is null (orchestration by that name doesn't exist)
if (factory == null) {
throw new IllegalStateException("No factory found for orchestrator: " + executionStarted.getName());
}
+ this.version = factory.getVersionName();
+
TaskOrchestration orchestrator = factory.create();
orchestrator.run(this);
break;
@@ -959,6 +1021,9 @@ private void processEvent(OrchestratorService.HistoryEvent e) {
case EXECUTIONTERMINATED:
this.handleExecutionTerminated(e);
break;
+ case EXECUTIONSTALLED:
+ this.logger.fine(() -> this.instanceId + ": Workflow execution stalled");
+ break;
case TASKSCHEDULED:
this.handleTaskScheduled(e);
break;
@@ -998,6 +1063,22 @@ private void processEvent(OrchestratorService.HistoryEvent e) {
}
}
+ public void setVersionNotRegistered() {
+ this.pendingActions.clear();
+
+ OrchestratorService.CompleteOrchestrationAction.Builder builder = OrchestratorService.CompleteOrchestrationAction
+ .newBuilder();
+ builder.setOrchestrationStatus(OrchestratorService.OrchestrationStatus.ORCHESTRATION_STATUS_STALLED);
+
+ int id = this.sequenceNumber++;
+ OrchestratorService.OrchestratorAction action = OrchestratorService.OrchestratorAction.newBuilder()
+ .setId(id)
+ .setCompleteOrchestration(builder.build())
+ .build();
+ this.pendingActions.put(id, action);
+
+ }
+
private class TaskRecord {
private final CompletableTask task;
private final String taskName;
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestratorResult.java b/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestratorResult.java
index 705a41d5c0..8243176031 100644
--- a/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestratorResult.java
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestratorResult.java
@@ -17,6 +17,7 @@
import java.util.Collection;
import java.util.Collections;
+import java.util.List;
final class TaskOrchestratorResult {
@@ -24,10 +25,16 @@ final class TaskOrchestratorResult {
private final String customStatus;
- public TaskOrchestratorResult(Collection actions, String customStatus) {
+ private final String version;
+
+ private final List patches;
+
+ public TaskOrchestratorResult(Collection actions,
+ String customStatus, String version, List patches) {
this.actions = Collections.unmodifiableCollection(actions);
- ;
this.customStatus = customStatus;
+ this.version = version;
+ this.patches = patches;
}
public Collection getActions() {
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/orchestration/TaskOrchestrationFactories.java b/durabletask-client/src/main/java/io/dapr/durabletask/orchestration/TaskOrchestrationFactories.java
new file mode 100644
index 0000000000..972cb838d4
--- /dev/null
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/orchestration/TaskOrchestrationFactories.java
@@ -0,0 +1,105 @@
+/*
+ * 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.durabletask.orchestration;
+
+import io.dapr.durabletask.orchestration.exception.VersionNotRegisteredException;
+
+import java.util.HashMap;
+
+public class TaskOrchestrationFactories {
+ final HashMap orchestrationFactories = new HashMap<>();
+ final HashMap> versionedOrchestrationFactories = new HashMap<>();
+ final HashMap latestVersionOrchestrationFactories = new HashMap<>();
+
+ /**
+ * Adds a new orchestration factory to the registry.
+ *
+ * @param factory the factory to add
+ */
+ public void addOrchestration(TaskOrchestrationFactory factory) {
+ String key = factory.getName();
+ if (this.emptyString(key)) {
+ throw new IllegalArgumentException("A non-empty task orchestration name is required.");
+ }
+
+ if (this.orchestrationFactories.containsKey(key)) {
+ throw new IllegalArgumentException(
+ String.format("A task orchestration factory named %s is already registered.", key));
+ }
+
+ if (emptyString(factory.getVersionName())) {
+ this.orchestrationFactories.put(key, factory);
+ return;
+ }
+
+ if (!this.versionedOrchestrationFactories.containsKey(key)) {
+ this.versionedOrchestrationFactories.put(key, new HashMap<>());
+ } else {
+ if (this.versionedOrchestrationFactories.get(key).containsKey(factory.getVersionName())) {
+ throw new IllegalArgumentException("The version name " + factory.getVersionName() + "for "
+ + factory.getName() + " is already registered.");
+ }
+ this.versionedOrchestrationFactories.get(key).put(factory.getVersionName(), factory);
+ if (factory.isLatestVersion()) {
+ this.latestVersionOrchestrationFactories.put(key, factory.getVersionName());
+ }
+ }
+ }
+
+ /**
+ * Gets the orchestration factory for the specified orchestration name.
+ *
+ * @param orchestrationName the orchestration name
+ * @return the orchestration factory
+ */
+ public TaskOrchestrationFactory getOrchestrationFactory(String orchestrationName) {
+ if (this.orchestrationFactories.containsKey(orchestrationName)) {
+ return this.orchestrationFactories.get(orchestrationName);
+ }
+
+ return this.getOrchestrationFactory(orchestrationName, "");
+ }
+
+ /**
+ * Gets the orchestration factory for the specified orchestration name and version.
+ *
+ * @param orchestrationName the orchestration name
+ * @param versionName the version name
+ * @return the orchestration factory
+ */
+ public TaskOrchestrationFactory getOrchestrationFactory(String orchestrationName, String versionName) {
+ if (this.orchestrationFactories.containsKey(orchestrationName)) {
+ return this.orchestrationFactories.get(orchestrationName);
+ }
+
+ if (!this.versionedOrchestrationFactories.containsKey(orchestrationName)) {
+ return null;
+ }
+
+ if (this.emptyString(versionName)) {
+ String latestVersion = this.latestVersionOrchestrationFactories.get(orchestrationName);
+ return this.versionedOrchestrationFactories.get(orchestrationName).get(latestVersion);
+ }
+
+ if (this.versionedOrchestrationFactories.get(orchestrationName).containsKey(versionName)) {
+ return this.versionedOrchestrationFactories.get(orchestrationName).get(versionName);
+ }
+
+ throw new VersionNotRegisteredException();
+ }
+
+ private boolean emptyString(String s) {
+ return s == null || s.isEmpty();
+ }
+}
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationFactory.java b/durabletask-client/src/main/java/io/dapr/durabletask/orchestration/TaskOrchestrationFactory.java
similarity index 87%
rename from durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationFactory.java
rename to durabletask-client/src/main/java/io/dapr/durabletask/orchestration/TaskOrchestrationFactory.java
index 274813b69f..a5e1b6a3cf 100644
--- a/durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationFactory.java
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/orchestration/TaskOrchestrationFactory.java
@@ -11,7 +11,9 @@
limitations under the License.
*/
-package io.dapr.durabletask;
+package io.dapr.durabletask.orchestration;
+
+import io.dapr.durabletask.TaskOrchestration;
/**
* Factory interface for producing {@link TaskOrchestration} implementations.
@@ -30,4 +32,8 @@ public interface TaskOrchestrationFactory {
* @return the created orchestration instance
*/
TaskOrchestration create();
+
+ String getVersionName();
+
+ Boolean isLatestVersion();
}
diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/orchestration/exception/VersionNotRegisteredException.java b/durabletask-client/src/main/java/io/dapr/durabletask/orchestration/exception/VersionNotRegisteredException.java
new file mode 100644
index 0000000000..f69ad9ea65
--- /dev/null
+++ b/durabletask-client/src/main/java/io/dapr/durabletask/orchestration/exception/VersionNotRegisteredException.java
@@ -0,0 +1,17 @@
+/*
+ * 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.orchestration.exception;
+
+public class VersionNotRegisteredException extends RuntimeException {
+}
diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/IntegrationTestBase.java b/durabletask-client/src/test/java/io/dapr/durabletask/IntegrationTestBase.java
index bbfcde0469..d0a8a8faa6 100644
--- a/durabletask-client/src/test/java/io/dapr/durabletask/IntegrationTestBase.java
+++ b/durabletask-client/src/test/java/io/dapr/durabletask/IntegrationTestBase.java
@@ -13,6 +13,7 @@
package io.dapr.durabletask;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactory;
import org.junit.jupiter.api.AfterEach;
import java.time.Duration;
@@ -67,6 +68,16 @@ public String getName() {
public TaskOrchestration create() {
return implementation;
}
+
+ @Override
+ public String getVersionName() {
+ return "";
+ }
+
+ @Override
+ public Boolean isLatestVersion() {
+ return false;
+ }
});
return this;
}
diff --git a/pom.xml b/pom.xml
index 37c6ecea7d..63596d0ed9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -53,7 +53,7 @@
2.0
1.21.3
- 3.4.9
+ 3.4.13
6.2.7
1.7.0
diff --git a/sdk-workflows/pom.xml b/sdk-workflows/pom.xml
index 7fd95807f1..6346811f25 100644
--- a/sdk-workflows/pom.xml
+++ b/sdk-workflows/pom.xml
@@ -22,6 +22,10 @@
dapr-sdk
${project.parent.version}