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; + } + } +}