From cf447a78dd9f4dc63c8f227ddbeb83b83d99330c Mon Sep 17 00:00:00 2001 From: shinseongsu Date: Fri, 30 Oct 2026 09:07:31 +0900 Subject: [PATCH] Include memo in test server list workflow executions The test server built WorkflowExecutionInfo for ListOpenWorkflowExecutions and ListClosedWorkflowExecutions from history without the memo, so callers always got an empty memo even though DescribeWorkflowExecution and the real server return it. Rebuild the memo from the started event and every WorkflowPropertiesModified upsert, using the same merge rules as the mutable state. Fixes #2480 --- CHANGELOG.md | 2 + .../testservice/TestWorkflowStoreImpl.java | 21 ++- .../ListWorkflowExecutionsMemoTest.java | 151 ++++++++++++++++++ 3 files changed, 173 insertions(+), 1 deletion(-) create mode 100644 temporal-test-server/src/test/java/io/temporal/testserver/functional/ListWorkflowExecutionsMemoTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index a390056d46..efda8dc6c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -93,6 +93,8 @@ to docs, or any other relevant information. ignored and retries were scheduled past them instead of failing with `RETRY_STATE_TIMEOUT`. - Worker heartbeats now list each plugin name once. Previously a plugin registered on both the `WorkflowClient` and the `WorkerFactory` was listed twice. +- Test server now includes the memo in `ListOpenWorkflowExecutions` and `ListClosedWorkflowExecutions` results, + including values upserted with `Workflow.upsertMemo`. Previously the memo was always empty there. ## Previous releases diff --git a/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowStoreImpl.java b/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowStoreImpl.java index 88c153ecf2..4ab0f36a6f 100644 --- a/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowStoreImpl.java +++ b/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowStoreImpl.java @@ -6,6 +6,8 @@ import com.google.protobuf.util.Timestamps; import io.grpc.Deadline; import io.grpc.Status; +import io.temporal.api.common.v1.Memo; +import io.temporal.api.common.v1.Payload; import io.temporal.api.common.v1.Priority; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.enums.v1.EventType; @@ -545,13 +547,30 @@ private WorkflowExecutionInfo constructWorkflowExecutionInfo( .setExecution(executionId.getExecution()) .setHistoryLength(history.size()) .setStartTime(history.get(0).getEventTime()) - .setType(history.get(0).getWorkflowExecutionStartedEventAttributes().getWorkflowType()); + .setType(history.get(0).getWorkflowExecutionStartedEventAttributes().getWorkflowType()) + .setMemo(memoFromHistory(history)); if (status != null) { info.setStatus(status); } return info.build(); } + private static Memo memoFromHistory(List history) { + // Only the last upsert of each key matters, so collect them and merge once. + Map upserts = new HashMap<>(); + for (HistoryEvent event : history) { + if (event.getEventType() == EventType.EVENT_TYPE_WORKFLOW_PROPERTIES_MODIFIED) { + upserts.putAll( + event.getWorkflowPropertiesModifiedEventAttributes().getUpsertedMemo().getFieldsMap()); + } + } + Map memo = + StateUtils.mergeMemo( + history.get(0).getWorkflowExecutionStartedEventAttributes().getMemo().getFieldsMap(), + upserts); + return Memo.newBuilder().putAllFields(memo).build(); + } + @Override public void close() { selfAdvancingTimer.shutdown(); diff --git a/temporal-test-server/src/test/java/io/temporal/testserver/functional/ListWorkflowExecutionsMemoTest.java b/temporal-test-server/src/test/java/io/temporal/testserver/functional/ListWorkflowExecutionsMemoTest.java new file mode 100644 index 0000000000..eb7ef551d2 --- /dev/null +++ b/temporal-test-server/src/test/java/io/temporal/testserver/functional/ListWorkflowExecutionsMemoTest.java @@ -0,0 +1,151 @@ +package io.temporal.testserver.functional; + +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.workflow.v1.WorkflowExecutionInfo; +import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.ListClosedWorkflowExecutionsRequest; +import io.temporal.api.workflowservice.v1.ListOpenWorkflowExecutionsRequest; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.SignalMethod; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.time.Duration; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; + +public class ListWorkflowExecutionsMemoTest { + private static final DataConverter DATA_CONVERTER = DefaultDataConverter.newDefaultInstance(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder().setWorkflowTypes(UpsertMemoWorkflowImpl.class).build(); + + @Test + public void listedExecutionsIncludeMemo() { + Map startMemo = new HashMap<>(); + startMemo.put("startKey", "startValue"); + startMemo.put("removedKey", "removedValue"); + WorkflowOptions options = + WorkflowOptions.newBuilder() + .setWorkflowId("list-memo-" + UUID.randomUUID()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setMemo(startMemo) + .build(); + UpsertMemoWorkflow workflow = + testWorkflowRule.getWorkflowClient().newWorkflowStub(UpsertMemoWorkflow.class, options); + WorkflowClient.start(workflow::execute); + WorkflowStub stub = WorkflowStub.fromTyped(workflow); + WorkflowExecution execution = stub.getExecution(); + + Map expectedMemo = new HashMap<>(); + expectedMemo.put("startKey", DATA_CONVERTER.toPayload("startValue").get()); + expectedMemo.put("upsertedKey", DATA_CONVERTER.toPayload("upsertedValue").get()); + + awaitMemoUpserted(execution); + WorkflowExecutionInfo open = findOpen(execution); + Assert.assertEquals(expectedMemo, open.getMemo().getFieldsMap()); + + workflow.finish(); + stub.getResult(Void.class); + WorkflowExecutionInfo closed = findClosed(execution); + Assert.assertEquals(expectedMemo, closed.getMemo().getFieldsMap()); + } + + private void awaitMemoUpserted(WorkflowExecution execution) { + DescribeWorkflowExecutionRequest request = + DescribeWorkflowExecutionRequest.newBuilder() + .setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace()) + .setExecution(execution) + .build(); + for (int i = 0; i < 50; i++) { + Map memo = + testWorkflowRule + .getWorkflowClient() + .getWorkflowServiceStubs() + .blockingStub() + .describeWorkflowExecution(request) + .getWorkflowExecutionInfo() + .getMemo() + .getFieldsMap(); + if (memo.containsKey("upsertedKey")) { + return; + } + testWorkflowRule.getTestEnvironment().sleep(Duration.ofMillis(100)); + } + throw new AssertionError("Memo was not upserted in time"); + } + + private WorkflowExecutionInfo findOpen(WorkflowExecution execution) { + List executions = + testWorkflowRule + .getWorkflowClient() + .getWorkflowServiceStubs() + .blockingStub() + .listOpenWorkflowExecutions( + ListOpenWorkflowExecutionsRequest.newBuilder() + .setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace()) + .build()) + .getExecutionsList(); + return findExecution(executions, execution); + } + + private WorkflowExecutionInfo findClosed(WorkflowExecution execution) { + List executions = + testWorkflowRule + .getWorkflowClient() + .getWorkflowServiceStubs() + .blockingStub() + .listClosedWorkflowExecutions( + ListClosedWorkflowExecutionsRequest.newBuilder() + .setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace()) + .build()) + .getExecutionsList(); + return findExecution(executions, execution); + } + + private static WorkflowExecutionInfo findExecution( + List executions, WorkflowExecution execution) { + return executions.stream() + .filter(info -> info.getExecution().getRunId().equals(execution.getRunId())) + .findFirst() + .orElseThrow(() -> new AssertionError("Execution not listed: " + execution)); + } + + @WorkflowInterface + public interface UpsertMemoWorkflow { + @WorkflowMethod + void execute(); + + @SignalMethod + void finish(); + } + + public static class UpsertMemoWorkflowImpl implements UpsertMemoWorkflow { + private boolean finished; + + @Override + public void execute() { + Workflow.upsertMemo(Collections.singletonMap("upsertedKey", "upsertedValue")); + Workflow.upsertMemo(Collections.singletonMap("removedKey", null)); + Workflow.await(() -> finished); + } + + @Override + public void finish() { + finished = true; + } + } +}