Add sdk name/version to task completions when unset or changed (#2422)

Author: Sushisource

temporal-sdk/src/main/java/io/temporal/internal/Config.java

@@ -24,7 +24,7 @@ public final class Config {
   private Config() {}
 
   /** Force new workflow task after workflow task timeout multiplied by this coefficient. */
-  public static final double WORKFLOW_TAK_HEARTBEAT_COEFFICIENT = 4d / 5d;
+  public static final double WORKFLOW_TASK_HEARTBEAT_COEFFICIENT = 4d / 5d;
 
   /**
    * Limit how many eager activities can be requested by the SDK in one workflow task completion

temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandler.java

@@ -146,7 +146,7 @@ public WorkflowTaskResult handleWorkflowTask(
           Deadline.after(
               (long)
                   (Durations.toNanos(startedEvent.getWorkflowTaskTimeout())
-                      * Config.WORKFLOW_TAK_HEARTBEAT_COEFFICIENT),
+                      * Config.WORKFLOW_TASK_HEARTBEAT_COEFFICIENT),
               TimeUnit.NANOSECONDS);
 
       if (workflowTask.getPreviousStartedEventId()
@@ -180,15 +180,23 @@ public WorkflowTaskResult handleWorkflowTask(
         throw context.getWorkflowTaskFailure();
       }
       Map<String, WorkflowQueryResult> queryResults = executeQueries(workflowTask.getQueriesMap());
-      return WorkflowTaskResult.newBuilder()
-          .setCommands(commands)
-          .setMessages(messages)
-          .setQueryResults(queryResults)
-          .setFinalCommand(context.isWorkflowMethodCompleted())
-          .setForceWorkflowTask(localActivityTaskCount > 0 && !context.isWorkflowMethodCompleted())
-          .setNonfirstLocalActivityAttempts(localActivityMeteringHelper.getNonfirstAttempts())
-          .setSdkFlags(newSdkFlags)
-          .build();
+      WorkflowTaskResult.Builder result =
+          WorkflowTaskResult.newBuilder()
+              .setCommands(commands)
+              .setMessages(messages)
+              .setQueryResults(queryResults)
+              .setFinalCommand(context.isWorkflowMethodCompleted())
+              .setForceWorkflowTask(
+                  localActivityTaskCount > 0 && !context.isWorkflowMethodCompleted())
+              .setNonfirstLocalActivityAttempts(localActivityMeteringHelper.getNonfirstAttempts())
+              .setSdkFlags(newSdkFlags);
+      if (workflowStateMachines.sdkNameToWrite() != null) {
+        result.setWriteSdkName(workflowStateMachines.sdkNameToWrite());
+      }
+      if (workflowStateMachines.sdkVersionToWrite() != null) {
+        result.setWriteSdkVersion(workflowStateMachines.sdkVersionToWrite());
+      }
+      return result.build();
     } finally {
       lock.unlock();
     }

temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java

@@ -247,12 +247,21 @@ private Result createCompletedWFTRequest(
       }
       completedRequest.setStickyAttributes(attributes);
     }
-    if (!result.getSdkFlags().isEmpty()) {
-      completedRequest =
-          completedRequest.setSdkMetadata(
-              WorkflowTaskCompletedMetadata.newBuilder()
-                  .addAllLangUsedFlags(result.getSdkFlags())
-                  .build());
+    List<Integer> sdkFlags = result.getSdkFlags();
+    String writeSdkName = result.getWriteSdkName();
+    String writeSdkVersion = result.getWriteSdkVersion();
+    if (!sdkFlags.isEmpty() || writeSdkName != null || writeSdkVersion != null) {
+      WorkflowTaskCompletedMetadata.Builder md = WorkflowTaskCompletedMetadata.newBuilder();
+      if (!sdkFlags.isEmpty()) {
+        md.addAllLangUsedFlags(sdkFlags);
+      }
+      if (writeSdkName != null) {
+        md.setSdkName(writeSdkName);
+      }
+      if (writeSdkVersion != null) {
+        md.setSdkVersion(writeSdkVersion);
+      }
+      completedRequest.setSdkMetadata(md.build());
     }
     return new Result(
         workflowType,

temporal-sdk/src/main/java/io/temporal/internal/replay/WorkflowTaskResult.java

@@ -41,6 +41,8 @@ public static final class Builder {
     private boolean forceWorkflowTask;
     private int nonfirstLocalActivityAttempts;
     private List<Integer> sdkFlags;
+    private String writeSdkName;
+    private String writeSdkVersion;
 
     public Builder setCommands(List<Command> commands) {
       this.commands = commands;
@@ -77,6 +79,16 @@ public Builder setSdkFlags(List<Integer> sdkFlags) {
       return this;
     }
 
+    public Builder setWriteSdkName(String writeSdkName) {
+      this.writeSdkName = writeSdkName;
+      return this;
+    }
+
+    public Builder setWriteSdkVersion(String writeSdkVersion) {
+      this.writeSdkVersion = writeSdkVersion;
+      return this;
+    }
+
     public WorkflowTaskResult build() {
       return new WorkflowTaskResult(
           commands == null ? Collections.emptyList() : commands,
@@ -85,7 +97,9 @@ public WorkflowTaskResult build() {
           finalCommand,
           forceWorkflowTask,
           nonfirstLocalActivityAttempts,
-          sdkFlags == null ? Collections.emptyList() : sdkFlags);
+          sdkFlags == null ? Collections.emptyList() : sdkFlags,
+          writeSdkName,
+          writeSdkVersion);
     }
   }
 
@@ -96,6 +110,8 @@ public WorkflowTaskResult build() {
   private final boolean forceWorkflowTask;
   private final int nonfirstLocalActivityAttempts;
   private final List<Integer> sdkFlags;
+  private final String writeSdkName;
+  private final String writeSdkVersion;
 
   private WorkflowTaskResult(
       List<Command> commands,
@@ -104,7 +120,9 @@ private WorkflowTaskResult(
       boolean finalCommand,
       boolean forceWorkflowTask,
       int nonfirstLocalActivityAttempts,
-      List<Integer> sdkFlags) {
+      List<Integer> sdkFlags,
+      String writeSdkName,
+      String writeSdkVersion) {
     this.commands = commands;
     this.messages = messages;
     this.nonfirstLocalActivityAttempts = nonfirstLocalActivityAttempts;
@@ -115,6 +133,8 @@ private WorkflowTaskResult(
     this.finalCommand = finalCommand;
     this.forceWorkflowTask = forceWorkflowTask;
     this.sdkFlags = sdkFlags;
+    this.writeSdkName = writeSdkName;
+    this.writeSdkVersion = writeSdkVersion;
   }
 
   public List<Command> getCommands() {
@@ -145,4 +165,12 @@ public int getNonfirstLocalActivityAttempts() {
   public List<Integer> getSdkFlags() {
     return sdkFlags;
   }
+
+  public String getWriteSdkName() {
+    return writeSdkName;
+  }
+
+  public String getWriteSdkVersion() {
+    return writeSdkVersion;
+  }
 }

temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java

@@ -43,11 +43,13 @@
 import io.temporal.internal.history.VersionMarkerUtils;
 import io.temporal.internal.sync.WorkflowThread;
 import io.temporal.internal.worker.LocalActivityResult;
+import io.temporal.serviceclient.Version;
 import io.temporal.worker.NonDeterministicException;
 import io.temporal.workflow.ChildWorkflowCancellationType;
 import io.temporal.workflow.Functions;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
 public final class WorkflowStateMachines {
@@ -179,6 +181,8 @@ enum HandleEventStatus {
   private final Set<String> acceptedUpdates = new HashSet<>();
 
   private final SdkFlags flags;
+  @Nonnull private String lastSeenSdkName = "";
+  @Nonnull private String lastSeenSdkVersion = "";
 
   public WorkflowStateMachines(
       StatesMachinesCallback callbacks, GetSystemInfoResponse.Capabilities capabilities) {
@@ -384,6 +388,12 @@ private void handleSingleEventLookahead(HistoryEvent event) {
           }
           flags.setSdkFlag(sdkFlag);
         }
+        if (!Strings.isNullOrEmpty(completedEvent.getSdkMetadata().getSdkName())) {
+          lastSeenSdkName = completedEvent.getSdkMetadata().getSdkName();
+        }
+        if (!Strings.isNullOrEmpty(completedEvent.getSdkMetadata().getSdkVersion())) {
+          lastSeenSdkVersion = completedEvent.getSdkMetadata().getSdkVersion();
+        }
         // Remove any finished update protocol state machines. We can't remove them on an event like
         // other state machines because a rejected update produces no event in history.
         protocolStateMachines.entrySet().removeIf(entry -> entry.getValue().isFinalState());
@@ -675,6 +685,26 @@ public EnumSet<SdkFlag> takeNewSdkFlags() {
     return flags.takeNewSdkFlags();
   }
 
+  /**
+   * @return If we need to write the SDK name upon WFT completion, return it
+   */
+  public String sdkNameToWrite() {
+    if (!lastSeenSdkName.equals(Version.SDK_NAME)) {
+      return Version.SDK_NAME;
+    }
+    return null;
+  }
+
+  /**
+   * @return If we need to write the SDK version upon WFT completion, return it
+   */
+  public String sdkVersionToWrite() {
+    if (!lastSeenSdkVersion.equals(Version.LIBRARY_VERSION)) {
+      return Version.LIBRARY_VERSION;
+    }
+    return null;
+  }
+
   private void prepareCommands() {
     if (preparing) {
       return;

temporal-sdk/src/test/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandlerTaskHandlerTests.java

@@ -43,6 +43,7 @@
 import io.temporal.internal.worker.WorkflowExecutorCache;
 import io.temporal.internal.worker.WorkflowRunLockManager;
 import io.temporal.internal.worker.WorkflowTaskHandler;
+import io.temporal.serviceclient.Version;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.testUtils.HistoryUtils;
 import io.temporal.testing.internal.SDKTestWorkflowRule;
@@ -208,6 +209,34 @@ public void ifStickyExecutionAttributesAreSetThenWorkflowsAreCached() throws Thr
     assertEquals(Durations.fromSeconds(5), attributes.getScheduleToStartTimeout());
   }
 
+  @Test
+  public void setsSdkNameAndVersionIfNotSetInHistory() throws Throwable {
+    assumeFalse("skipping for docker tests", SDKTestWorkflowRule.useExternalService);
+
+    WorkflowExecutorCache cache =
+        new WorkflowExecutorCache(10, new WorkflowRunLockManager(), new NoopScope());
+    WorkflowTaskHandler taskHandler =
+        new ReplayWorkflowTaskHandler(
+            "namespace",
+            setUpMockWorkflowFactory(),
+            cache,
+            SingleWorkerOptions.newBuilder().build(),
+            InternalUtils.createStickyTaskQueue("sticky", "taskQueue"),
+            Duration.ofSeconds(5),
+            testWorkflowRule.getWorkflowServiceStubs(),
+            null);
+
+    PollWorkflowTaskQueueResponse workflowTask =
+        HistoryUtils.generateWorkflowTaskWithInitialHistory();
+
+    WorkflowTaskHandler.Result result = taskHandler.handleWorkflowTask(workflowTask);
+
+    assertTrue(result.isCompletionCommand());
+    assertEquals(Version.SDK_NAME, result.getTaskCompleted().getSdkMetadata().getSdkName());
+    assertEquals(
+        Version.LIBRARY_VERSION, result.getTaskCompleted().getSdkMetadata().getSdkVersion());
+  }
+
   private ReplayWorkflowFactory setUpMockWorkflowFactory() throws Throwable {
     ReplayWorkflow mockWorkflow = mock(ReplayWorkflow.class);
     ReplayWorkflowFactory mockFactory = mock(ReplayWorkflowFactory.class);

temporal-sdk/src/test/java/io/temporal/internal/statemachines/TestHistoryBuilder.java

@@ -616,6 +616,10 @@ public long getPreviousStartedEventId() {
     return previousStartedEventId;
   }
 
+  public long getWorkflowTaskScheduledEventId() {
+    return workflowTaskScheduledEventId;
+  }
+
   public long getWorkflowTaskStartedEventId() {
     return workflowTaskScheduledEventId + 1;
   }

temporal-sdk/src/test/java/io/temporal/internal/statemachines/WorkflowStateMachinesTest.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.temporal.internal.statemachines;
+
+import static org.junit.Assert.assertEquals;
+
+import io.temporal.api.enums.v1.EventType;
+import io.temporal.api.history.v1.WorkflowTaskCompletedEventAttributes;
+import io.temporal.api.sdk.v1.WorkflowTaskCompletedMetadata;
+import io.temporal.serviceclient.Version;
+import java.util.Optional;
+import org.junit.Test;
+
+public class WorkflowStateMachinesTest {
+  private WorkflowStateMachines stateMachines;
+
+  private WorkflowStateMachines newStateMachines(TestEntityManagerListenerBase listener) {
+    return new WorkflowStateMachines(listener, m -> {});
+  }
+
+  private class TestActivityListener extends TestEntityManagerListenerBase {
+    @Override
+    public void buildWorkflow(AsyncWorkflowBuilder<Void> builder) {
+      builder.add(v -> stateMachines.completeWorkflow(Optional.empty()));
+    }
+  }
+
+  private void sdkNameAndVersionTest(
+      String inputSdkVersion,
+      String inputSdkName,
+      String expectedSdkName,
+      String expectedSdkVersion) {
+    TestHistoryBuilder h = new TestHistoryBuilder();
+    TestEntityManagerListenerBase listener = new TestActivityListener();
+    stateMachines = newStateMachines(listener);
+
+    h.add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED);
+    h.addWorkflowTaskScheduledAndStarted();
+    h.add(
+        EventType.EVENT_TYPE_WORKFLOW_TASK_COMPLETED,
+        WorkflowTaskCompletedEventAttributes.newBuilder()
+            .setScheduledEventId(h.getWorkflowTaskScheduledEventId())
+            .setSdkMetadata(
+                WorkflowTaskCompletedMetadata.newBuilder()
+                    .setSdkVersion(inputSdkVersion)
+                    .setSdkName(inputSdkName)));
+    h.addWorkflowTaskScheduledAndStarted();
+    assertEquals(2, h.getWorkflowTaskCount());
+
+    h.handleWorkflowTaskTakeCommands(stateMachines, 2);
+
+    assertEquals(expectedSdkName, stateMachines.sdkNameToWrite());
+    assertEquals(expectedSdkVersion, stateMachines.sdkVersionToWrite());
+  }
+
+  @Test
+  public void testWritesSdkNameAndVersionWhenDifferent() {
+    sdkNameAndVersionTest("hi", "skflajk", Version.SDK_NAME, Version.LIBRARY_VERSION);
+  }
+
+  @Test
+  public void doesNotWriteSdkNameAndVersionWhenSame() {
+    sdkNameAndVersionTest(Version.LIBRARY_VERSION, Version.SDK_NAME, null, null);
+  }
+
+  @Test
+  public void writesOnlyNameIfChanged() {
+    sdkNameAndVersionTest(Version.LIBRARY_VERSION, "sakflasjklf", Version.SDK_NAME, null);
+  }
+
+  @Test
+  public void writesOnlyVersionIfChanged() {
+    sdkNameAndVersionTest("safklasjf", Version.SDK_NAME, null, Version.LIBRARY_VERSION);
+  }
+}