Deserialized Search Attributes API and full Test Server implementation for Search Attributes (#1067)

Author: Spikhalskiy

temporal-kotlin/build.gradle

@@ -7,7 +7,7 @@ buildscript {
 plugins {
 //    id 'org.jetbrains.kotlin.jvm' version '1.4.32'
 //    id 'org.jetbrains.kotlin.jvm' version '1.5.32'
-//    id 'org.jetbrains.kotlin.jvm' version '1.6.10'
+//    id 'org.jetbrains.kotlin.jvm' version '1.6.20'
     id 'org.jetbrains.kotlin.jvm' version "${kotlinVersion}"
     id 'org.jlleitschuh.gradle.ktlint' version '10.2.1'
 }

temporal-sdk/src/main/java/io/temporal/client/WorkflowOptions.java

@@ -27,6 +27,7 @@
 import io.temporal.common.context.ContextPropagator;
 import io.temporal.internal.common.OptionsUtils;
 import java.time.Duration;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import javax.annotation.Nullable;
@@ -92,7 +93,7 @@ public static final class Builder {
 
     private Map<String, Object> memo;
 
-    private Map<String, Object> searchAttributes;
+    private Map<String, ?> searchAttributes;
 
     private List<ContextPropagator> contextPropagators;
 
@@ -211,10 +212,26 @@ public Builder setMemo(Map<String, Object> memo) {
     }
 
     /**
-     * Specifies additional indexed information in result of list workflow. The type of value should
-     * be basic type such as: String, Integer, Boolean, Double，LocalDateTime
+     * Specifies Search Attributes map {@code searchAttributes} that will be attached to the
+     * Workflow. Search Attributes are additional indexed information attributed to workflow and
+     * used for search and visibility.
+     *
+     * <p>The search attributes can be used in query of List/Scan/Count workflow APIs. The key and
+     * its value type must be registered on Temporal server side.
+     *
+     * <p>Supported Java types of the value:
+     *
+     * <ul>
+     *   <li>String
+     *   <li>Long, Integer, Short, Byte
+     *   <li>Boolean
+     *   <li>Double
+     *   <li>OffsetDateTime
+     *   <li>{@link Collection} of the types above
+     * </ul>
      */
-    public Builder setSearchAttributes(Map<String, Object> searchAttributes) {
+    // Workflow#upsertSearchAttributes docs needs to be kept in sync with this method
+    public Builder setSearchAttributes(Map<String, ?> searchAttributes) {
       this.searchAttributes = searchAttributes;
       return this;
     }
@@ -285,7 +302,7 @@ public WorkflowOptions validateBuildWithDefaults() {
 
   private final Map<String, Object> memo;
 
-  private final Map<String, Object> searchAttributes;
+  private final Map<String, ?> searchAttributes;
 
   private final List<ContextPropagator> contextPropagators;
 
@@ -299,7 +316,7 @@ private WorkflowOptions(
       RetryOptions retryOptions,
       String cronSchedule,
       Map<String, Object> memo,
-      Map<String, Object> searchAttributes,
+      Map<String, ?> searchAttributes,
       List<ContextPropagator> contextPropagators) {
     this.workflowId = workflowId;
     this.workflowIdReusePolicy = workflowIdReusePolicy;
@@ -350,7 +367,7 @@ public Map<String, Object> getMemo() {
     return memo;
   }
 
-  public Map<String, Object> getSearchAttributes() {
+  public Map<String, ?> getSearchAttributes() {
     return searchAttributes;
   }
 

…nternal/common/SearchAttributesUtil.java …/io/temporal/common/SearchAttribute.javatemporal-sdk/src/test/java/io/temporal/internal/common/SearchAttributesUtil.java renamed to temporal-sdk/src/main/java/io/temporal/common/SearchAttribute.java temporal-sdk/src/test/java/io/temporal/internal/common/SearchAttributesUtil.java renamed to temporal-sdk/src/main/java/io/temporal/common/SearchAttribute.java

@@ -17,20 +17,17 @@
  *  permissions and limitations under the License.
  */
 
-package io.temporal.internal.common;
+package io.temporal.common;
 
-import io.temporal.api.common.v1.SearchAttributes;
-import io.temporal.common.converter.DataConverter;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
 
-public class SearchAttributesUtil {
-  private static final DataConverter jsonConverter = DataConverter.getDefaultInstance();
-
-  public static <T> T getValueFromSearchAttributes(
-      SearchAttributes searchAttributes, String key, Class<T> classType) {
-    if (searchAttributes == null || key == null || key.isEmpty()) {
-      return null;
-    }
-    return jsonConverter.fromPayload(
-        searchAttributes.getIndexedFieldsOrThrow(key), classType, classType);
-  }
+public class SearchAttribute {
+  /**
+   * Passing this value as a search attribute value into {@link
+   * io.temporal.workflow.Workflow#upsertSearchAttributes(Map)} will lead to unsetting the search
+   * attribute with the corresponded name if any present.
+   */
+  public static final List<Object> UNSET_VALUE = Collections.emptyList();
 }

temporal-sdk/src/main/java/io/temporal/common/converter/ByteArrayPayloadConverter.java

@@ -43,7 +43,6 @@ public Optional<Payload> toData(Object value) throws DataConverterException {
   }
 
   @Override
-  @SuppressWarnings("unchecked")
   public <T> T fromData(Payload content, Class<T> valueClass, Type valueType)
       throws DataConverterException {
     ByteString data = content.getData();

temporal-sdk/src/main/java/io/temporal/common/converter/JacksonJsonPayloadConverter.java

@@ -22,6 +22,7 @@
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.PropertyAccessor;
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
@@ -39,6 +40,10 @@ public class JacksonJsonPayloadConverter implements PayloadConverter {
 
   public JacksonJsonPayloadConverter() {
     mapper = new ObjectMapper();
+    // preserve the original value of timezone coming from the server in Payload
+    // without adjusting to the host timezone
+    // may be important if the replay is happening on a host in another timezone
+    mapper.configure(DeserializationFeature.ADJUST_DATES_TO_CONTEXT_TIME_ZONE, false);
     mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
     mapper.registerModule(new JavaTimeModule());
     mapper.registerModule(new Jdk8Module());

temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptor.java

@@ -486,7 +486,7 @@ <R> R mutableSideEffect(
 
   UUID randomUUID();
 
-  void upsertSearchAttributes(Map<String, Object> searchAttributes);
+  void upsertSearchAttributes(Map<String, ?> searchAttributes);
 
   /**
    * Intercepts creation of the workflow child thread.

temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptorBase.java

@@ -135,7 +135,7 @@ public UUID randomUUID() {
   }
 
   @Override
-  public void upsertSearchAttributes(Map<String, Object> searchAttributes) {
+  public void upsertSearchAttributes(Map<String, ?> searchAttributes) {
     next.upsertSearchAttributes(searchAttributes);
   }
 

temporal-sdk/src/main/java/io/temporal/failure/ServerFailure.java

@@ -19,11 +19,17 @@
 
 package io.temporal.failure;
 
+import javax.annotation.Nullable;
+
 /** Exceptions originated at the Temporal service. */
 public final class ServerFailure extends TemporalFailure {
   private final boolean nonRetryable;
 
-  public ServerFailure(String message, boolean nonRetryable, Throwable cause) {
+  public ServerFailure(String message, boolean nonRetryable) {
+    this(message, nonRetryable, null);
+  }
+
+  public ServerFailure(String message, boolean nonRetryable, @Nullable Throwable cause) {
     super(message, message, cause);
     this.nonRetryable = nonRetryable;
   }

temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientHelper.java

@@ -35,9 +35,9 @@
 import io.temporal.client.WorkflowOptions;
 import io.temporal.common.RetryOptions;
 import io.temporal.common.context.ContextPropagator;
-import io.temporal.common.converter.DataConverter;
 import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
 import io.temporal.internal.common.ProtobufTimeUtils;
+import io.temporal.internal.common.SearchAttributesUtil;
 import java.util.*;
 
 final class RootWorkflowClientHelper {
@@ -91,12 +91,8 @@ StartWorkflowExecutionRequest newStartWorkflowExecutionRequest(
           Memo.newBuilder()
               .putAllFields(intoPayloadMap(clientOptions.getDataConverter(), options.getMemo())));
     }
-    if (options.getSearchAttributes() != null) {
-      request.setSearchAttributes(
-          SearchAttributes.newBuilder()
-              .putAllIndexedFields(
-                  intoPayloadMap(
-                      DataConverter.getDefaultInstance(), options.getSearchAttributes())));
+    if (options.getSearchAttributes() != null && !options.getSearchAttributes().isEmpty()) {
+      request.setSearchAttributes(SearchAttributesUtil.encode(options.getSearchAttributes()));
     }
 
     Header grpcHeader =

temporal-sdk/src/main/java/io/temporal/internal/common/HistoryJsonUtils.java

@@ -19,7 +19,6 @@
 
 package io.temporal.internal.common;
 
-import com.google.common.base.CaseFormat;
 import com.jayway.jsonpath.Configuration;
 import com.jayway.jsonpath.DocumentContext;
 import com.jayway.jsonpath.JsonPath;
@@ -30,10 +29,6 @@
  * Helper methods supporting transformation of History's "Proto Json" compatible format, which is
  * supported by {@link com.google.protobuf.util.JsonFormat} to the format of Temporal history
  * supported by tctl and back.
- *
- * @see <a
- *     href="https://github.com/temporalio/gogo-protobuf/commit/b38fb010909b8f81e2e600dc6f04925fc71d6a5e">
- *     Related commit to Go Proto module</>
  */
 class HistoryJsonUtils {
   private static final Configuration JSON_PATH_CONFIGURATION =
@@ -63,11 +58,11 @@ private enum EnumValueConversionPolicy {
   }
 
   public static String protoJsonToHistoryFormatJson(String protoJson) {
-    return convertEnumValues(protoJson, HistoryJsonUtils::enumProtoToHistory);
+    return convertEnumValues(protoJson, ProtoEnumNameUtils::uniqueToSimplifiedName);
   }
 
   public static String historyFormatJsonToProtoJson(String historyFormatJson) {
-    return convertEnumValues(historyFormatJson, HistoryJsonUtils::enumHistoryToProto);
+    return convertEnumValues(historyFormatJson, ProtoEnumNameUtils::simplifiedToUniqueName);
   }
 
   private static String convertEnumValues(
@@ -83,26 +78,4 @@ private static String convertEnumValues(
     }
     return parsed.jsonString();
   }
-
-  private static String enumProtoToHistory(String protoEnumValue, String prefix) {
-    if (!protoEnumValue.startsWith(prefix)) {
-      throw new IllegalArgumentException("protoEnumValue should start with " + prefix + " prefix");
-    }
-    protoEnumValue = protoEnumValue.substring(prefix.length());
-    return screamingCaseEventTypeToCamelCase(protoEnumValue);
-  }
-
-  private static String enumHistoryToProto(String historyEnumValue, String prefix) {
-    return prefix + camelCaseToScreamingCase(historyEnumValue);
-  }
-
-  // https://github.com/temporalio/gogo-protobuf/commit/b38fb010909b8f81e2e600dc6f04925fc71d6a5e
-  private static String camelCaseToScreamingCase(String camel) {
-    return CaseFormat.UPPER_CAMEL.converterTo(CaseFormat.UPPER_UNDERSCORE).convert(camel);
-  }
-
-  // https://github.com/temporalio/gogo-protobuf/commit/b38fb010909b8f81e2e600dc6f04925fc71d6a5e
-  private static String screamingCaseEventTypeToCamelCase(String screaming) {
-    return CaseFormat.UPPER_UNDERSCORE.converterTo(CaseFormat.UPPER_CAMEL).convert(screaming);
-  }
 }