From c67e0f01aa6c2cccdadaf29cff888f190f647773 Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Tue, 18 Aug 2026 16:39:59 -0400 Subject: [PATCH] feature(extstore): integrate into workflow worker pipeline, including replay handler. --- .../replay/ReplayWorkflowTaskHandler.java | 10 +- .../ServiceWorkflowHistoryIterator.java | 21 +++- .../internal/worker/WorkflowWorker.java | 100 ++++++++++++++-- .../ServiceWorkflowHistoryIteratorTest.java | 113 ++++++++++++++++++ .../internal/worker/WorkflowWorkerTest.java | 62 ++++++++++ 5 files changed, 293 insertions(+), 13 deletions(-) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java index f5b7cb0d29..a9505c372b 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java @@ -23,6 +23,7 @@ import io.temporal.common.converter.DataConverter; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.WorkflowExecutionUtils; +import io.temporal.internal.payload.storage.ExternalStorage; import io.temporal.internal.worker.*; import io.temporal.payload.context.WorkflowSerializationContext; import io.temporal.serviceclient.MetricsTag; @@ -77,6 +78,12 @@ public WorkflowTaskHandler.Result handleWorkflowTask(PollWorkflowTaskQueueRespon String workflowType = workflowTask.getWorkflowType().getName(); Scope metricsScope = options.getMetricsScope().tagged(ImmutableMap.of(MetricsTag.WORKFLOW_TYPE, workflowType)); + ExternalStorage externalStorage = options.getExternalStorage(); + if (externalStorage == null) { + ExternalStorage.throwIfContainsReference(workflowTask); + } else { + workflowTask = externalStorage.retrieveBlocking(workflowTask); + } return handleWorkflowTaskWithQuery(workflowTask.toBuilder(), metricsScope); } @@ -94,7 +101,8 @@ private Result handleWorkflowTaskWithQuery( logWorkflowTaskToBeProcessed(workflowTask, createdNew); ServiceWorkflowHistoryIterator historyIterator = - new ServiceWorkflowHistoryIterator(service, namespace, workflowTask, metricsScope); + new ServiceWorkflowHistoryIterator( + service, namespace, workflowTask, metricsScope, options.getExternalStorage()); boolean finalCommand; Result result; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ServiceWorkflowHistoryIterator.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ServiceWorkflowHistoryIterator.java index 229b66186e..aa72485423 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ServiceWorkflowHistoryIterator.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/replay/ServiceWorkflowHistoryIterator.java @@ -12,12 +12,14 @@ import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryRequest; import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse; import io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueResponseOrBuilder; +import io.temporal.internal.payload.storage.ExternalStorage; import io.temporal.internal.retryer.GrpcRetryer; import io.temporal.serviceclient.RpcRetryOptions; import io.temporal.serviceclient.WorkflowServiceStubs; import java.time.Duration; import java.util.Iterator; import java.util.NoSuchElementException; +import javax.annotation.Nullable; /** Supports iteration over history while loading new pages through calls to the service. */ class ServiceWorkflowHistoryIterator implements WorkflowHistoryIterator { @@ -29,6 +31,7 @@ class ServiceWorkflowHistoryIterator implements WorkflowHistoryIterator { private final Scope metricsScope; private final PollWorkflowTaskQueueResponseOrBuilder task; private final GrpcRetryer grpcRetryer; + private final @Nullable ExternalStorage externalStorage; private Deadline deadline; private Iterator current; ByteString nextPageToken; @@ -38,10 +41,20 @@ class ServiceWorkflowHistoryIterator implements WorkflowHistoryIterator { String namespace, PollWorkflowTaskQueueResponseOrBuilder task, Scope metricsScope) { + this(service, namespace, task, metricsScope, null); + } + + ServiceWorkflowHistoryIterator( + WorkflowServiceStubs service, + String namespace, + PollWorkflowTaskQueueResponseOrBuilder task, + Scope metricsScope, + @Nullable ExternalStorage externalStorage) { this.service = service; this.namespace = namespace; this.task = task; this.metricsScope = metricsScope; + this.externalStorage = externalStorage; // TODO Refactor WorkflowHistoryIteratorTest or WorkflowHistoryIterator to remove this check. // `service == null` shouldn't be allowed as it's needed for a normal functioning of this // class. @@ -64,7 +77,13 @@ public boolean hasNext() { // true. GetWorkflowExecutionHistoryResponse response = queryWorkflowExecutionHistory(); - current = response.getHistory().getEventsList().iterator(); + History history = response.getHistory(); + if (externalStorage == null) { + ExternalStorage.throwIfContainsReference(history); + } else { + history = externalStorage.retrieveBlocking(history); + } + current = history.getEventsList().iterator(); nextPageToken = response.getNextPageToken(); // Server can return an empty page, but a valid nextPageToken that contains // more events. diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java index 3eed1099d3..2d625beca8 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java @@ -6,10 +6,14 @@ import com.google.common.base.Preconditions; import com.google.common.base.Strings; import com.google.protobuf.ByteString; +import com.google.protobuf.MessageOrBuilder; import com.uber.m3.tally.Scope; import com.uber.m3.tally.Stopwatch; import com.uber.m3.util.ImmutableMap; import io.grpc.StatusRuntimeException; +import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributesOrBuilder; +import io.temporal.api.command.v1.SignalExternalWorkflowExecutionCommandAttributesOrBuilder; +import io.temporal.api.command.v1.StartChildWorkflowExecutionCommandAttributesOrBuilder; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.enums.v1.QueryResultType; import io.temporal.api.enums.v1.TaskQueueKind; @@ -18,9 +22,14 @@ import io.temporal.api.workflowservice.v1.*; import io.temporal.failure.ApplicationFailure; import io.temporal.internal.logging.LoggerTag; +import io.temporal.internal.payload.storage.ExternalStorage; +import io.temporal.internal.payload.visitor.MessageVisitor; import io.temporal.internal.retryer.GrpcMessageTooLargeException; import io.temporal.internal.retryer.GrpcRetryer; import io.temporal.payload.context.WorkflowSerializationContext; +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverTargetInfo; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.RpcRetryOptions; import io.temporal.serviceclient.WorkflowServiceStubs; @@ -381,6 +390,55 @@ public String toString() { options.getIdentity(), namespace, taskQueue); } + private T storeOutboundPayloads( + T request, @Nullable StorageDriverTargetInfo target) { + ExternalStorage externalStorage = options.getExternalStorage(); + return externalStorage == null ? request : externalStorage.storeBlocking(request, target); + } + + private T storeOutboundPayloads( + T request, + @Nullable StorageDriverTargetInfo target, + MessageVisitor targetVisitor) { + ExternalStorage externalStorage = options.getExternalStorage(); + return externalStorage == null + ? request + : externalStorage.storeBlocking(request, target, targetVisitor); + } + + @Nullable + private StorageDriverTargetInfo workflowStorageTarget( + WorkflowExecution execution, String workflowType) { + if (options.getExternalStorage() == null) { + return null; + } + return new StorageDriverWorkflowInfo( + namespace, execution.getWorkflowId(), execution.getRunId(), workflowType); + } + + static StorageDriverTargetInfo refineStorageTarget( + String namespace, StorageDriverTargetInfo current, MessageOrBuilder message) { + if (message instanceof ScheduleActivityTaskCommandAttributesOrBuilder) { + ScheduleActivityTaskCommandAttributesOrBuilder attrs = + (ScheduleActivityTaskCommandAttributesOrBuilder) message; + return new StorageDriverActivityInfo( + namespace, attrs.getActivityId(), null, attrs.getActivityType().getName()); + } + if (message instanceof StartChildWorkflowExecutionCommandAttributesOrBuilder) { + StartChildWorkflowExecutionCommandAttributesOrBuilder attrs = + (StartChildWorkflowExecutionCommandAttributesOrBuilder) message; + return new StorageDriverWorkflowInfo( + namespace, attrs.getWorkflowId(), null, attrs.getWorkflowType().getName()); + } + if (message instanceof SignalExternalWorkflowExecutionCommandAttributesOrBuilder) { + WorkflowExecution execution = + ((SignalExternalWorkflowExecutionCommandAttributesOrBuilder) message).getExecution(); + return new StorageDriverWorkflowInfo( + namespace, execution.getWorkflowId(), execution.getRunId(), null); + } + return current; + } + private class TaskHandlerImpl implements PollTaskExecutor.TaskHandler { final WorkflowTaskHandler handler; @@ -453,7 +511,10 @@ public void handle(WorkflowTask task) throws Exception { if (queryCompleted != null) { try { sendDirectQueryCompletedResponse( - currentTask.getTaskToken(), queryCompleted.toBuilder(), workflowTypeScope); + currentTask.getTaskToken(), + queryCompleted.toBuilder(), + workflowTypeScope, + workflowStorageTarget(workflowExecution, workflowType)); } catch (StatusRuntimeException e) { GrpcMessageTooLargeException tooLargeException = GrpcMessageTooLargeException.tryWrap(e); @@ -473,7 +534,10 @@ public void handle(WorkflowTask task) throws Exception { .setErrorMessage(failure.getMessage()) .setFailure(failure); sendDirectQueryCompletedResponse( - currentTask.getTaskToken(), queryFailedBuilder, workflowTypeScope); + currentTask.getTaskToken(), + queryFailedBuilder, + workflowTypeScope, + workflowStorageTarget(workflowExecution, workflowType)); } } else { try { @@ -489,7 +553,8 @@ public void handle(WorkflowTask task) throws Exception { currentTask.getTaskToken(), requestBuilder, result.getRequestRetryOptions(), - workflowTypeScope); + workflowTypeScope, + workflowStorageTarget(workflowExecution, workflowType)); // If we were processing a speculative WFT the server may instruct us that the // task was dropped by resting out event ID. long resetEventId = response.getResetHistoryEventId(); @@ -509,7 +574,8 @@ public void handle(WorkflowTask task) throws Exception { currentTask.getTaskToken(), taskFailed.toBuilder(), result.getRequestRetryOptions(), - workflowTypeScope); + workflowTypeScope, + workflowStorageTarget(workflowExecution, workflowType)); } // Apply post-completion metrics only if runnable present and the above succeeded @@ -546,7 +612,8 @@ public void handle(WorkflowTask task) throws Exception { currentTask.getTaskToken(), taskFailedBuilder, result.getRequestRetryOptions(), - workflowTypeScope); + workflowTypeScope, + workflowStorageTarget(workflowExecution, workflowType)); } } } catch (Exception e) { @@ -651,7 +718,8 @@ private RespondWorkflowTaskCompletedResponse sendTaskCompleted( ByteString taskToken, RespondWorkflowTaskCompletedRequest.Builder taskCompleted, RpcRetryOptions retryOptions, - Scope workflowTypeMetricsScope) { + Scope workflowTypeMetricsScope, + @Nullable StorageDriverTargetInfo storageTarget) { GrpcRetryer.GrpcRetryerOptions grpcRetryOptions = new GrpcRetryer.GrpcRetryerOptions( RpcRetryOptions.newBuilder().buildWithDefaultsFrom(retryOptions), null); @@ -674,12 +742,16 @@ private RespondWorkflowTaskCompletedResponse sendTaskCompleted( taskCompleted.setBinaryChecksum(options.getBuildId()); } + MessageVisitor storageTargetVisitor = + (current, message) -> refineStorageTarget(namespace, current, message); + RespondWorkflowTaskCompletedRequest request = + storeOutboundPayloads(taskCompleted.build(), storageTarget, storageTargetVisitor); return grpcRetryer.retryWithResult( () -> service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, workflowTypeMetricsScope) - .respondWorkflowTaskCompleted(taskCompleted.build()), + .respondWorkflowTaskCompleted(request), grpcRetryOptions); } @@ -688,7 +760,8 @@ private void sendTaskFailed( ByteString taskToken, RespondWorkflowTaskFailedRequest.Builder taskFailed, RpcRetryOptions retryOptions, - Scope workflowTypeMetricsScope) { + Scope workflowTypeMetricsScope, + @Nullable StorageDriverTargetInfo storageTarget) { GrpcRetryer.GrpcRetryerOptions grpcRetryOptions = new GrpcRetryer.GrpcRetryerOptions( RpcRetryOptions.newBuilder().buildWithDefaultsFrom(retryOptions), null); @@ -702,25 +775,30 @@ private void sendTaskFailed( taskFailed.setWorkerVersion(options.workerVersionStamp()); } + RespondWorkflowTaskFailedRequest request = + storeOutboundPayloads(taskFailed.build(), storageTarget); grpcRetryer.retry( () -> service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, workflowTypeMetricsScope) - .respondWorkflowTaskFailed(taskFailed.build()), + .respondWorkflowTaskFailed(request), grpcRetryOptions); } private void sendDirectQueryCompletedResponse( ByteString taskToken, RespondQueryTaskCompletedRequest.Builder queryCompleted, - Scope workflowTypeMetricsScope) { + Scope workflowTypeMetricsScope, + @Nullable StorageDriverTargetInfo storageTarget) { queryCompleted.setTaskToken(taskToken).setNamespace(namespace); + RespondQueryTaskCompletedRequest request = + storeOutboundPayloads(queryCompleted.build(), storageTarget); // Do not retry query response service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, workflowTypeMetricsScope) - .respondQueryTaskCompleted(queryCompleted.build()); + .respondQueryTaskCompleted(request); } private void logExceptionDuringResultReporting( diff --git a/temporal-sdk/src/test/java/io/temporal/internal/replay/ServiceWorkflowHistoryIteratorTest.java b/temporal-sdk/src/test/java/io/temporal/internal/replay/ServiceWorkflowHistoryIteratorTest.java index ad0c665800..12063d3bc2 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/replay/ServiceWorkflowHistoryIteratorTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/replay/ServiceWorkflowHistoryIteratorTest.java @@ -1,12 +1,29 @@ package io.temporal.internal.replay; import com.google.protobuf.ByteString; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; import io.temporal.api.history.v1.History; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.history.v1.WorkflowExecutionStartedEventAttributes; import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse; import io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueResponse; +import io.temporal.internal.payload.storage.ExternalStorage; +import io.temporal.internal.payload.storage.ExternalStorageNotConfiguredException; +import io.temporal.payload.storage.ExternalStorageOptions; +import io.temporal.payload.storage.StorageDriver; +import io.temporal.payload.storage.StorageDriverClaim; +import io.temporal.payload.storage.StorageDriverRetrieveContext; +import io.temporal.payload.storage.StorageDriverStoreContext; import io.temporal.testUtils.HistoryUtils; import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.NoSuchElementException; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Assert; import org.junit.Test; @@ -84,4 +101,100 @@ GetWorkflowExecutionHistoryResponse queryWorkflowExecutionHistory() { Assert.assertThrows(NoSuchElementException.class, iterator::next); Assert.assertEquals(4, timesCalledServer.get()); } + + @Test + public void resolvesExternalStorageReferencesInFetchedPages() { + ExternalStorage storage = inMemoryStorage(); + History inline = historyWithInput(payload("big-input")); + History stored = storage.storeBlocking(inline, null); + Assert.assertNotEquals( + "stored history should hold a reference, not the inline payload", inline, stored); + + ServiceWorkflowHistoryIterator iterator = fetchingIterator(stored, storage); + + HistoryEvent event = iterator.next(); + Assert.assertEquals( + payload("big-input"), + event.getWorkflowExecutionStartedEventAttributes().getInput().getPayloads(0)); + } + + @Test + public void failsLoudWhenAFetchedPageHasAReferenceAndStorageIsNotConfigured() { + History stored = inMemoryStorage().storeBlocking(historyWithInput(payload("big-input")), null); + + ServiceWorkflowHistoryIterator iterator = fetchingIterator(stored, null); + + Assert.assertThrows(ExternalStorageNotConfiguredException.class, iterator::hasNext); + } + + private static ServiceWorkflowHistoryIterator fetchingIterator( + History page, ExternalStorage storage) { + PollWorkflowTaskQueueResponse workflowTask = + PollWorkflowTaskQueueResponse.newBuilder().setNextPageToken(NEXT_PAGE_TOKEN).build(); + return new ServiceWorkflowHistoryIterator(null, "default", workflowTask, null, storage) { + @Override + GetWorkflowExecutionHistoryResponse queryWorkflowExecutionHistory() { + return GetWorkflowExecutionHistoryResponse.newBuilder().setHistory(page).build(); + } + }; + } + + private static ExternalStorage inMemoryStorage() { + return ExternalStorage.create( + ExternalStorageOptions.newBuilder() + .setDriver(new InMemoryDriver()) + .setPayloadSizeThreshold(0) + .build()); + } + + private static History historyWithInput(Payload payload) { + return History.newBuilder() + .addEvents( + HistoryEvent.newBuilder() + .setWorkflowExecutionStartedEventAttributes( + WorkflowExecutionStartedEventAttributes.newBuilder() + .setInput(Payloads.newBuilder().addPayloads(payload)))) + .build(); + } + + private static Payload payload(String data) { + return Payload.newBuilder().setData(ByteString.copyFromUtf8(data)).build(); + } + + private static final class InMemoryDriver implements StorageDriver { + private final Map objects = new HashMap<>(); + private int counter = 0; + + @Override + public String getName() { + return "test"; + } + + @Override + public String getType() { + return "test.inmemory"; + } + + @Override + public synchronized CompletableFuture> store( + StorageDriverStoreContext context, List payloads) { + List claims = new ArrayList<>(); + for (Payload payload : payloads) { + String key = "k-" + (counter++); + objects.put(key, payload); + claims.add(new StorageDriverClaim(Collections.singletonMap("key", key))); + } + return CompletableFuture.completedFuture(claims); + } + + @Override + public synchronized CompletableFuture> retrieve( + StorageDriverRetrieveContext context, List claims) { + List payloads = new ArrayList<>(); + for (StorageDriverClaim claim : claims) { + payloads.add(objects.get(claim.getClaimData().get("key"))); + } + return CompletableFuture.completedFuture(payloads); + } + } } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java index 5cd1fc8d3e..c97d85a188 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java @@ -12,6 +12,11 @@ import com.uber.m3.tally.RootScopeBuilder; import com.uber.m3.tally.Scope; import com.uber.m3.util.ImmutableMap; +import io.temporal.api.command.v1.CompleteWorkflowExecutionCommandAttributes; +import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributes; +import io.temporal.api.command.v1.SignalExternalWorkflowExecutionCommandAttributes; +import io.temporal.api.command.v1.StartChildWorkflowExecutionCommandAttributes; +import io.temporal.api.common.v1.ActivityType; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.common.v1.WorkflowType; import io.temporal.api.workflowservice.v1.*; @@ -20,6 +25,9 @@ import io.temporal.internal.replay.ReplayWorkflow; import io.temporal.internal.replay.ReplayWorkflowFactory; import io.temporal.internal.replay.ReplayWorkflowTaskHandler; +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverTargetInfo; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.testUtils.Eventually; import io.temporal.testUtils.HistoryUtils; @@ -448,4 +456,58 @@ private ReplayWorkflowFactory setUpMockWorkflowFactory() throws Throwable { when(mockWorkflow.eventLoop()).thenReturn(false); return mockFactory; } + + @Test + public void refineStorageTargetPointsActivityCommandsAtTheActivity() { + StorageDriverTargetInfo workflowDefault = + new StorageDriverWorkflowInfo("ns", "wf-1", "run-1", "MyWorkflow"); + ScheduleActivityTaskCommandAttributes command = + ScheduleActivityTaskCommandAttributes.newBuilder() + .setActivityId("act-1") + .setActivityType(ActivityType.newBuilder().setName("MyActivity")) + .build(); + + assertEquals( + new StorageDriverActivityInfo("ns", "act-1", null, "MyActivity"), + WorkflowWorker.refineStorageTarget("ns", workflowDefault, command)); + } + + @Test + public void refineStorageTargetPointsChildWorkflowCommandsAtTheChild() { + StorageDriverTargetInfo parent = + new StorageDriverWorkflowInfo("ns", "parent", "parent-run", "Parent"); + StartChildWorkflowExecutionCommandAttributes command = + StartChildWorkflowExecutionCommandAttributes.newBuilder() + .setWorkflowId("child-1") + .setWorkflowType(WorkflowType.newBuilder().setName("Child")) + .build(); + + assertEquals( + new StorageDriverWorkflowInfo("ns", "child-1", null, "Child"), + WorkflowWorker.refineStorageTarget("ns", parent, command)); + } + + @Test + public void refineStorageTargetPointsSignalCommandsAtTheTargetWorkflow() { + StorageDriverTargetInfo self = new StorageDriverWorkflowInfo("ns", "self", "self-run", "Self"); + SignalExternalWorkflowExecutionCommandAttributes command = + SignalExternalWorkflowExecutionCommandAttributes.newBuilder() + .setExecution( + WorkflowExecution.newBuilder().setWorkflowId("other").setRunId("other-run")) + .build(); + + assertEquals( + new StorageDriverWorkflowInfo("ns", "other", "other-run", null), + WorkflowWorker.refineStorageTarget("ns", self, command)); + } + + @Test + public void refineStorageTargetKeepsTheCurrentTargetForOtherCommands() { + StorageDriverTargetInfo current = + new StorageDriverWorkflowInfo("ns", "wf-1", "run-1", "MyWorkflow"); + CompleteWorkflowExecutionCommandAttributes command = + CompleteWorkflowExecutionCommandAttributes.newBuilder().build(); + + assertSame(current, WorkflowWorker.refineStorageTarget("ns", current, command)); + } }