Skip to content

exstore/pipeline integration #2978

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cconstable wants to merge 22 commits into from
Open

exstore/pipeline integration #2978

Show file tree
Hide file tree
Changes from 17 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
815c3a3
feat(exstore): add to client and worker options.
cconstable Jul 29, 2026
bb346a9
feat(extstore): integrate external storage into client and worker pip…
cconstable Jul 29, 2026
c17310c
feat(extstore): offload activity completion and heartbeat payloads
cconstable Jul 29, 2026
d9a9dd2
feat(extstore): offload nexus worker task payloads
cconstable Jul 29, 2026
3d08ed5
feat(extstore): offload remaining client RPCs and WFT query/failure p…
cconstable Jul 29, 2026
ce4cfd1
add clientWithoutExternalStorageReadingAReferenceFails test
cconstable Jul 29, 2026
30534e1
fix(extstore): resolve external-storage references during offline replay
cconstable Jul 30, 2026
029158d
fix(extstore): release Nexus slots on retrieval failure
cconstable Jul 30, 2026
55d9e9c
fix(extstore): preserve missing storage failures
cconstable Jul 30, 2026
f12e8d9
fix(extstore): correct activity storage targets.
cconstable Jul 30, 2026
cfa0550
fix(extstore): handle manual completion storage failures
cconstable Jul 30, 2026
4e09350
fix(extstore): bound client I/O and honor interrupts.
cconstable Jul 30, 2026
35eca09
fix(extstore): add multi-operation workflow target
cconstable Jul 30, 2026
4a30a20
fix(extstore): restore integration surface on *Transformer after rebase
cconstable Aug 10, 2026
0ed1b46
fix comments and tests
cconstable Aug 10, 2026
48c2cdd
refactor(extstore): remove special casing extstore errors. rename tra…
cconstable Aug 10, 2026
466bec5
refactor: rename await in message transformer to getOrCancel. seems m…
cconstable Aug 11, 2026
a44b6a6
fix: add @Nullable annotations to some extstore interfaces.
cconstable Aug 12, 2026
4d907c8
refactor(extstore): move payload visit concurrency onto ExternalStora…
cconstable Aug 12, 2026
a040feb
refactor: rename ExternalStorageMessageTransformer -> ExternalStorage
cconstable Aug 12, 2026
dc6454d
fix: use message type detection for detecting external payloads.
cconstable Aug 13, 2026
7f80874
address PR feedback before splitting into separate prs.
cconstable Aug 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,15 @@
import io.temporal.internal.WorkflowThreadMarker;
import io.temporal.internal.client.*;
import io.temporal.internal.client.NexusStartWorkflowResponse;
import io.temporal.internal.client.external.ExternalStorageGenericWorkflowClient;
import io.temporal.internal.client.external.GenericWorkflowClient;
import io.temporal.internal.client.external.GenericWorkflowClientImpl;
import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
import io.temporal.internal.common.PluginUtils;
import io.temporal.internal.payload.storage.ExternalStorageMessageTransformer;
import io.temporal.internal.sync.StubMarker;
import io.temporal.internal.worker.HeartbeatManager;
import io.temporal.payload.storage.ExternalStorageOptions;
import io.temporal.serviceclient.MetricsTag;
import io.temporal.serviceclient.WorkflowServiceStubs;
import io.temporal.serviceclient.WorkflowServiceStubsPlugin;
Expand Down Expand Up @@ -106,15 +109,28 @@ public static WorkflowClient newInstance(
.getOptions()
.getMetricsScope()
.tagged(MetricsTag.defaultTags(options.getNamespace()));
this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
ExternalStorageOptions externalStorageOptions = options.getExternalStorage();
ExternalStorageMessageTransformer externalStorage =

@jmaeagle99 jmaeagle99 Aug 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: I would move this to inside the if-then a few lines below and update the condition to check if externalStorageOptions is not null.

externalStorageOptions == null
? null
: ExternalStorageMessageTransformer.create(externalStorageOptions);
GenericWorkflowClient genericClient =
new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
if (externalStorage != null) {
genericClient =
new ExternalStorageGenericWorkflowClient(
genericClient, externalStorage, options.getNamespace());
}
this.genericClient = genericClient;
this.interceptors = options.getInterceptors();
this.workflowClientCallsInvoker = initializeClientInvoker();
this.manualActivityCompletionClientFactory =
ManualActivityCompletionClientFactory.newFactory(
workflowServiceStubs,
options.getNamespace(),
options.getIdentity(),
options.getDataConverter());
options.getDataConverter(),
externalStorage);

java.time.Duration heartbeatInterval = options.getWorkerHeartbeatInterval();
if (!heartbeatInterval.isNegative()) {
Expand Down
40 changes: 35 additions & 5 deletions temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import io.temporal.common.converter.DataConverter;
import io.temporal.common.converter.GlobalDataConverter;
import io.temporal.common.interceptors.WorkflowClientInterceptor;
import io.temporal.payload.storage.ExternalStorageOptions;
import java.lang.management.ManagementFactory;
import java.time.Duration;
import java.util.Arrays;
Expand Down Expand Up @@ -52,6 +53,7 @@ public static final class Builder {
private QueryRejectCondition queryRejectCondition;
private WorkflowClientPlugin[] plugins;
private Duration workerHeartbeatInterval;
private ExternalStorageOptions externalStorage;

private Builder() {}

Expand All @@ -68,6 +70,7 @@ private Builder(WorkflowClientOptions options) {
queryRejectCondition = options.queryRejectCondition;
plugins = options.plugins;
workerHeartbeatInterval = options.workerHeartbeatInterval;
externalStorage = options.externalStorage;
}

public Builder setNamespace(String namespace) {
Expand Down Expand Up @@ -170,6 +173,17 @@ public Builder setWorkerHeartbeatInterval(Duration workerHeartbeatInterval) {
return this;
}

/**
* Configures offloading of large payloads to external storage for workflows and activities
* created through this client and its workers. When null (the default), external storage is
* disabled.
*/
@Experimental
public Builder setExternalStorage(ExternalStorageOptions externalStorage) {
this.externalStorage = externalStorage;
return this;
}

public WorkflowClientOptions build() {
return new WorkflowClientOptions(
namespace,
Expand All @@ -180,7 +194,8 @@ public WorkflowClientOptions build() {
contextPropagators,
queryRejectCondition,
plugins == null ? EMPTY_PLUGINS : plugins,
resolveHeartbeatInterval(workerHeartbeatInterval));
resolveHeartbeatInterval(workerHeartbeatInterval),
externalStorage);
}

/**
Expand All @@ -207,7 +222,8 @@ public WorkflowClientOptions validateAndBuildWithDefaults() {
? QueryRejectCondition.QUERY_REJECT_CONDITION_UNSPECIFIED
: queryRejectCondition,
plugins == null ? EMPTY_PLUGINS : plugins,
resolveHeartbeatInterval(workerHeartbeatInterval));
resolveHeartbeatInterval(workerHeartbeatInterval),
externalStorage);
}

private static Duration resolveHeartbeatInterval(Duration raw) {
Expand Down Expand Up @@ -250,6 +266,8 @@ private static Duration resolveHeartbeatInterval(Duration raw) {

private final Duration workerHeartbeatInterval;

private final ExternalStorageOptions externalStorage;

private WorkflowClientOptions(
String namespace,
DataConverter dataConverter,
Expand All @@ -259,7 +277,8 @@ private WorkflowClientOptions(
List<ContextPropagator> contextPropagators,
QueryRejectCondition queryRejectCondition,
WorkflowClientPlugin[] plugins,
Duration workerHeartbeatInterval) {
Duration workerHeartbeatInterval,
ExternalStorageOptions externalStorage) {
this.namespace = namespace;
this.dataConverter = dataConverter;
this.interceptors = interceptors;
Expand All @@ -269,6 +288,7 @@ private WorkflowClientOptions(
this.queryRejectCondition = queryRejectCondition;
this.plugins = plugins;
this.workerHeartbeatInterval = workerHeartbeatInterval;
this.externalStorage = externalStorage;
}

/**
Expand Down Expand Up @@ -335,6 +355,12 @@ public Duration getWorkerHeartbeatInterval() {
return workerHeartbeatInterval;
}

/** External storage configuration, or null when external storage is disabled. */
@Experimental
public ExternalStorageOptions getExternalStorage() {
return externalStorage;
}

@Override
public String toString() {
return "WorkflowClientOptions{"
Expand All @@ -359,6 +385,8 @@ public String toString() {
+ Arrays.toString(plugins)
+ ", workerHeartbeatInterval="
+ workerHeartbeatInterval
+ ", externalStorage="
+ externalStorage
+ '}';
}

Expand All @@ -376,7 +404,8 @@ public boolean equals(Object o) {
&& queryRejectCondition == that.queryRejectCondition
&& Arrays.equals(plugins, that.plugins)
&& com.google.common.base.Objects.equal(
workerHeartbeatInterval, that.workerHeartbeatInterval);
workerHeartbeatInterval, that.workerHeartbeatInterval)
&& com.google.common.base.Objects.equal(externalStorage, that.externalStorage);
}

@Override
Expand All @@ -390,6 +419,7 @@ public int hashCode() {
contextPropagators,
queryRejectCondition,
Arrays.hashCode(plugins),
workerHeartbeatInterval);
workerHeartbeatInterval,
externalStorage);
}
}
5 changes: 5 additions & 0 deletions temporal-sdk/src/main/java/io/temporal/common/converter/PayloadAndFailureDataConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import io.temporal.api.common.v1.Payloads;
import io.temporal.api.failure.v1.Failure;
import io.temporal.failure.DefaultFailureConverter;
import io.temporal.internal.payload.storage.ExternalStorageNotConfiguredException;
import io.temporal.payload.context.SerializationContext;
import java.lang.reflect.Type;
import java.util.*;
Expand Down Expand Up @@ -71,6 +72,10 @@ public <T> T fromPayload(Payload payload, Class<T> valueClass, Type valueType)
return (T) new RawValue(payload);
}

if (payload.getExternalPayloadsCount() > 0) {

@jmaeagle99 jmaeagle99 Aug 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Doesn't this need to check if external storage is not configured AND there are externally stored payloads?

@jmaeagle99 jmaeagle99 Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, it shouldn't be checking for the external_payloads field on the payload, it should be checking the metadata of the payload, probably using ExternalStorageReferences.tryParseReference or similar.

@cconstable cconstable Aug 13, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it should be checking the metadata of the payload

yes, updated

Doesn't this need to check if external storage is not configured AND there are externally stored payloads?

No because if external storage is configured, it will have already walked the messages and resolved the references before getting here. If no driver can retrieve the message that error should be thrown before this point. This is just a single place to catch "external storage was not configured and we have a reference payload still".

Ideally, I'd really like to remove external storage from this class but that either requires doing a bunch of eager message walking up front or adding lazy checks to a bunch of random places in the sdk which seems error prone and architecturally unsound.

throw new ExternalStorageNotConfiguredException();
}

try {
String encoding =
payload.getMetadataOrThrow(EncodingKeys.METADATA_ENCODING_KEY).toString(UTF_8);
Expand Down
11 changes: 8 additions & 3 deletions ...-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextFactoryImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import io.temporal.client.WorkflowClient;
import io.temporal.common.converter.DataConverter;
import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
import io.temporal.internal.payload.storage.ExternalStorageMessageTransformer;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.Arrays;
Expand All @@ -21,6 +22,7 @@ public class ActivityExecutionContextFactoryImpl implements ActivityExecutionCon
private final DataConverter dataConverter;
private final ScheduledExecutorService heartbeatExecutor;
private final ManualActivityCompletionClientFactory manualCompletionClientFactory;
private final ExternalStorageMessageTransformer externalStorage;
private final ConcurrentMap<ByteBuffer, ActivityExecutionContextImpl> activeContexts =
new ConcurrentHashMap<>();

Expand All @@ -31,7 +33,8 @@ public ActivityExecutionContextFactoryImpl(
Duration maxHeartbeatThrottleInterval,
Duration defaultHeartbeatThrottleInterval,
DataConverter dataConverter,
ScheduledExecutorService heartbeatExecutor) {
ScheduledExecutorService heartbeatExecutor,
ExternalStorageMessageTransformer externalStorage) {
this.client = Objects.requireNonNull(client);
this.identity = identity;
this.namespace = Objects.requireNonNull(namespace);
Expand All @@ -40,9 +43,10 @@ public ActivityExecutionContextFactoryImpl(
Objects.requireNonNull(defaultHeartbeatThrottleInterval);
this.dataConverter = Objects.requireNonNull(dataConverter);
this.heartbeatExecutor = Objects.requireNonNull(heartbeatExecutor);
this.externalStorage = externalStorage;
this.manualCompletionClientFactory =
ManualActivityCompletionClientFactory.newFactory(
client.getWorkflowServiceStubs(), namespace, identity, dataConverter);
client.getWorkflowServiceStubs(), namespace, identity, dataConverter, externalStorage);
}

@Override
Expand All @@ -63,7 +67,8 @@ public InternalActivityExecutionContext createContext(
identity,
maxHeartbeatThrottleInterval,
defaultHeartbeatThrottleInterval,
() -> cleanupContext(info.getTaskToken(), false));
() -> cleanupContext(info.getTaskToken(), false),
externalStorage);
activeContexts.put(taskToken, context);
return context;
}
Expand Down
18 changes: 15 additions & 3 deletions temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,17 @@
import io.temporal.common.CancellationToken;
import io.temporal.common.converter.DataConverter;
import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
import io.temporal.internal.payload.storage.ExternalStorageMessageTransformer;
import io.temporal.payload.context.ActivitySerializationContext;
import io.temporal.payload.storage.StorageDriverActivityInfo;
import io.temporal.workflow.Functions;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

/**
Expand Down Expand Up @@ -55,7 +58,8 @@ class ActivityExecutionContextImpl implements InternalActivityExecutionContext {
String identity,
Duration maxHeartbeatThrottleInterval,
Duration defaultHeartbeatThrottleInterval,
Functions.Proc closeCallback) {
Functions.Proc closeCallback,
@Nullable ExternalStorageMessageTransformer externalStorage) {
this.client = client;
this.activity = activity;
this.metricsScope = metricsScope;
Expand All @@ -73,7 +77,8 @@ class ActivityExecutionContextImpl implements InternalActivityExecutionContext {
metricsScope,
identity,
maxHeartbeatThrottleInterval,
defaultHeartbeatThrottleInterval);
defaultHeartbeatThrottleInterval,
externalStorage);
}

/**
Expand Down Expand Up @@ -155,7 +160,14 @@ public ManualActivityCompletionClient useLocalManualCompletion() {
new ActivitySerializationContext(info);
return new CompletionAwareManualCompletionClient(
manualCompletionClientFactory.getClient(
info.getTaskToken(), metricsScope, activitySerializationContext),
info.getTaskToken(),
metricsScope,
activitySerializationContext,
new StorageDriverActivityInfo(
info.getNamespace(),
info.getActivityId(),
info.getActivityRunId(),
info.getActivityType())),
completionHandle);
} finally {
lock.unlock();
Expand Down
Loading
Loading