Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<HistoryEvent> history) {
// Only the last upsert of each key matters, so collect them and merge once.
Map<String, Payload> upserts = new HashMap<>();
for (HistoryEvent event : history) {
if (event.getEventType() == EventType.EVENT_TYPE_WORKFLOW_PROPERTIES_MODIFIED) {
upserts.putAll(
event.getWorkflowPropertiesModifiedEventAttributes().getUpsertedMemo().getFieldsMap());
}
}
Map<String, Payload> memo =
StateUtils.mergeMemo(
history.get(0).getWorkflowExecutionStartedEventAttributes().getMemo().getFieldsMap(),
upserts);
return Memo.newBuilder().putAllFields(memo).build();
}

@Override
public void close() {
selfAdvancingTimer.shutdown();
Expand Down
Original file line number Diff line number Diff line change
@@ -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<String, Object> 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<String, Payload> 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<String, Payload> 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<WorkflowExecutionInfo> 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<WorkflowExecutionInfo> executions =
testWorkflowRule
.getWorkflowClient()
.getWorkflowServiceStubs()
.blockingStub()
.listClosedWorkflowExecutions(
ListClosedWorkflowExecutionsRequest.newBuilder()
.setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace())
.build())
.getExecutionsList();
return findExecution(executions, execution);
}

private static WorkflowExecutionInfo findExecution(
List<WorkflowExecutionInfo> 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;
}
}
}
Loading