Feature version header (#226)

* Added LibraryVersion and LibraryFeature headers

* Added full history loading when sticky task is received for a workflow not in cache

Author: mfateev

build.gradle

@@ -68,6 +68,25 @@ compileJava {
             ".*/generated-sources/.*" << "-Werror"
 }
 
+// Generation version.properties for value to be included into the request header
+task createProperties(dependsOn: processResources) {
+    doLast {
+        def subdir = new File("$buildDir/resources/main/com/uber/cadence/")
+        if( !subdir.exists() ) {
+            subdir.mkdirs()
+        }
+        new File("$buildDir/resources/main/com/uber/cadence/version.properties").withWriter { w ->
+            Properties p = new Properties()
+            p['cadence-client-version'] = project.version.toString()
+            p.store w, null
+        }
+    }
+}
+
+classes {
+    dependsOn createProperties
+}
+
 compileTestJava {
     options.encoding = 'UTF-8'
     options.compilerArgs << "-Xlint:unchecked" << "-Xlint:deprecation" << "-XepExcludedPaths:" +

src/main/java/com/uber/cadence/internal/Version.java

@@ -0,0 +1,71 @@
+/*
+ *  Copyright 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 file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file 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 com.uber.cadence.internal;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Properties;
+
+/**
+ * below are the metadata which will be embedded as part of headers in every rpc call made by this
+ * client to cadence server.
+ *
+ * <p>Update to the metadata below is typically done by the cadence team as part of a major feature
+ * or behavior change
+ */
+public class Version {
+
+  /**
+   * Library Version is a semver that represents the version of this cadence client library. This
+   * represent API changes visibile to Cadence client side library consumers. I.e. developers that
+   * are writing workflows. So every time we change API that can affect them we have to change this
+   * number. Format: MAJOR.MINOR.PATCH
+   */
+  public static final String LIBRARY_VERSION;
+
+  /**
+   * Feature Version is a semver that represents the feature set of this cadence client library
+   * support. This can be used for client capibility check, on Cadence server, for backward
+   * compatibility Format: MAJOR.MINOR.PATCH
+   */
+  public static final String FEATURE_VERSION = "1.0.0";
+
+  static {
+    // Load version from version.properties generated by Gradle into build/resources/main directory.
+    Properties prop = new Properties();
+    InputStream in = Version.class.getResourceAsStream("/com/uber/cadence/version.properties");
+    if (in == null) {
+      LIBRARY_VERSION = "UNKNOWN";
+    } else {
+      String version = null;
+      try {
+        try {
+          prop.load(in);
+          version = prop.getProperty("cadence-client-version");
+        } finally {
+          in.close();
+        }
+      } catch (IOException e) {
+        if (version == null) {
+          version = "UNKNOWN";
+        }
+      }
+      LIBRARY_VERSION = version;
+    }
+  }
+}

src/main/java/com/uber/cadence/internal/replay/DeciderCache.java

@@ -18,9 +18,12 @@
 package com.uber.cadence.internal.replay;
 
 import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
+import com.google.common.util.concurrent.ExecutionError;
+import com.google.common.util.concurrent.UncheckedExecutionException;
 import com.uber.cadence.PollForDecisionTaskResponse;
 import com.uber.cadence.internal.common.ThrowableFunc1;
 import com.uber.cadence.internal.metrics.MetricsType;
@@ -30,6 +33,7 @@
 import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import org.slf4j.Logger;
@@ -75,18 +79,26 @@ public Decider getOrCreate(
       invalidate(decisionTask);
       return cache.get(runId, () -> createReplayDecider.apply(decisionTask));
     }
-    return getUnchecked(runId);
-  }
-
-  public Decider getUnchecked(String runId) throws Exception {
+    AtomicBoolean miss = new AtomicBoolean();
+    Decider result = null;
     try {
-      Decider cachedDecider = cache.getUnchecked(runId);
-      metricsScope.counter(MetricsType.STICKY_CACHE_HIT).inc(1);
-      return cachedDecider;
-    } catch (CacheLoader.InvalidCacheLoadException e) {
-      metricsScope.counter(MetricsType.STICKY_CACHE_MISS).inc(1);
-      throw new EvictedException(runId);
+      result =
+          cache.get(
+              runId,
+              () -> {
+                miss.set(true);
+                return createReplayDecider.apply(decisionTask);
+              });
+    } catch (UncheckedExecutionException | ExecutionError e) {
+      Throwables.throwIfUnchecked(e.getCause());
+    } finally {
+      if (miss.get()) {
+        metricsScope.counter(MetricsType.STICKY_CACHE_MISS).inc(1);
+      } else {
+        metricsScope.counter(MetricsType.STICKY_CACHE_HIT).inc(1);
+      }
     }
+    return result;
   }
 
   public void evictAny(String runId) throws InterruptedException {

src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java

@@ -210,6 +210,18 @@ public boolean isAnyTypeSupported() {
 
   private Decider createDecider(PollForDecisionTaskResponse decisionTask) throws Exception {
     WorkflowType workflowType = decisionTask.getWorkflowType();
+    List<HistoryEvent> events = decisionTask.getHistory().getEvents();
+    // Sticky decision task with partial history
+    if (events.isEmpty() || events.get(0).getEventId() > 1) {
+      GetWorkflowExecutionHistoryRequest getHistoryRequest =
+          new GetWorkflowExecutionHistoryRequest()
+              .setDomain(domain)
+              .setExecution(decisionTask.getWorkflowExecution());
+      GetWorkflowExecutionHistoryResponse getHistoryResponse =
+          service.GetWorkflowExecutionHistory(getHistoryRequest);
+      decisionTask.setHistory(getHistoryResponse.getHistory());
+      decisionTask.setNextPageToken(getHistoryResponse.getNextPageToken());
+    }
     DecisionsHelper decisionsHelper = new DecisionsHelper(decisionTask);
     ReplayWorkflow workflow = workflowFactory.getWorkflow(workflowType);
     return new ReplayDecider(

src/main/java/com/uber/cadence/serviceclient/WorkflowServiceTChannel.java

@@ -74,6 +74,7 @@
 import com.uber.cadence.WorkflowExecutionAlreadyStartedError;
 import com.uber.cadence.WorkflowService;
 import com.uber.cadence.WorkflowService.GetWorkflowExecutionHistory_result;
+import com.uber.cadence.internal.Version;
 import com.uber.cadence.internal.common.CheckedExceptionWrapper;
 import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.internal.metrics.NoopScope;
@@ -164,7 +165,7 @@ private ClientOptions(Builder builder) {
       }
       this.metricsScope = builder.metricsScope;
       if (builder.transportHeaders != null) {
-        this.transportHeaders = builder.transportHeaders;
+        this.transportHeaders = ImmutableMap.copyOf(builder.transportHeaders);
       } else {
         this.transportHeaders = ImmutableMap.of();
       }
@@ -358,7 +359,13 @@ public WorkflowServiceTChannel(String host, int port, ClientOptions options) {
       ArrayList<InetSocketAddress> peers = new ArrayList<>();
       peers.add(new InetSocketAddress(address, port));
       this.subChannel = tChannel.makeSubChannel(options.getServiceName()).setPeers(peers);
-      log.info("Initialized TChannel for service " + this.subChannel.getServiceName());
+      log.info(
+          "Initialized TChannel for service "
+              + this.subChannel.getServiceName()
+              + ", LibraryVersion: "
+              + Version.LIBRARY_VERSION
+              + ", FeatureVersion: "
+              + Version.FEATURE_VERSION);
     } catch (UnknownHostException e) {
       tChannel.shutdown();
       throw new RuntimeException("Unable to get name of host " + host, e);
@@ -391,6 +398,8 @@ private static Map<String, String> getThriftHeaders() {
     return ImmutableMap.<String, String>builder()
         .put("user-name", envUserName)
         .put("host-name", envHostname)
+        .put("cadence-client-library-version", Version.LIBRARY_VERSION)
+        .put("cadence-client-feature-version", Version.FEATURE_VERSION)
         .build();
   }
 

src/test/java/com/uber/cadence/internal/replay/ReplayDeciderCacheTests.java

@@ -18,14 +18,14 @@
 package com.uber.cadence.internal.replay;
 
 import static junit.framework.TestCase.assertEquals;
-import static org.junit.Assert.assertNotSame;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.*;
 import static org.mockito.Matchers.anyInt;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.*;
 
 import com.uber.cadence.HistoryEvent;
 import com.uber.cadence.PollForDecisionTaskResponse;
+import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowQuery;
 import com.uber.cadence.internal.metrics.MetricsTag;
 import com.uber.cadence.internal.metrics.MetricsType;
@@ -61,29 +61,39 @@ public void whenHistoryIsFullNewReplayDeciderIsReturnedAndCached_InitiallyEmpty(
     Decider decider = replayDeciderCache.getOrCreate(decisionTask, this::createFakeDecider);
 
     // Assert
-    assertEquals(decider, replayDeciderCache.getUnchecked(runId));
+    assertNotEquals(decider, replayDeciderCache.getOrCreate(decisionTask, this::createFakeDecider));
   }
 
   @Test
   public void whenHistoryIsFullNewReplayDeciderIsReturned_InitiallyCached() throws Exception {
+    TestWorkflowService service = new TestWorkflowService();
+
     // Arrange
     DeciderCache replayDeciderCache = new DeciderCache(10, NoopScope.getInstance());
-    PollForDecisionTaskResponse decisionTask =
-        HistoryUtils.generateDecisionTaskWithInitialHistory();
+    PollForDecisionTaskResponse decisionTask1 =
+        HistoryUtils.generateDecisionTaskWithInitialHistory(
+            "domain", "taskList", "workflowType", service);
 
-    String runId = decisionTask.getWorkflowExecution().getRunId();
-    Decider decider = replayDeciderCache.getOrCreate(decisionTask, this::createFakeDecider);
-    assertEquals(decider, replayDeciderCache.getUnchecked(runId));
+    String runId = decisionTask1.getWorkflowExecution().getRunId();
+    Decider decider = replayDeciderCache.getOrCreate(decisionTask1, this::createFakeDecider);
+
+    PollForDecisionTaskResponse decisionTask2 =
+        HistoryUtils.generateDecisionTaskWithPartialHistoryFromExistingTask(
+            decisionTask1, "domain", "stickyTaskList", service);
+
+    assertEquals(
+        decider, replayDeciderCache.getOrCreate(decisionTask2, this::doNotCreateFakeDecider));
 
     // Act
-    Decider decider2 = replayDeciderCache.getOrCreate(decisionTask, this::createFakeDecider);
+    Decider decider2 = replayDeciderCache.getOrCreate(decisionTask2, this::createFakeDecider);
 
     // Assert
-    assertEquals(decider2, replayDeciderCache.getUnchecked(runId));
-    assertNotSame(decider2, decider);
+    assertEquals(decider2, replayDeciderCache.getOrCreate(decisionTask2, this::createFakeDecider));
+    assertSame(decider2, decider);
+    service.close();
   }
 
-  @Test
+  @Test(timeout = 2000)
   public void whenHistoryIsPartialCachedEntryIsReturned() throws Exception {
     // Arrange
     Map<String, String> tags =
@@ -97,31 +107,30 @@ public void whenHistoryIsPartialCachedEntryIsReturned() throws Exception {
 
     DeciderCache replayDeciderCache = new DeciderCache(10, scope);
     TestWorkflowService service = new TestWorkflowService();
+    service.lockTimeSkipping("test");
     PollForDecisionTaskResponse decisionTask =
         HistoryUtils.generateDecisionTaskWithInitialHistory(
             "domain", "taskList", "workflowType", service);
 
     String runId = decisionTask.getWorkflowExecution().getRunId();
     Decider decider = replayDeciderCache.getOrCreate(decisionTask, this::createFakeDecider);
-    assertEquals(decider, replayDeciderCache.getUnchecked(runId));
 
     // Act
     PollForDecisionTaskResponse decisionTask2 =
         HistoryUtils.generateDecisionTaskWithPartialHistoryFromExistingTask(
             decisionTask, "domain", "stickyTaskList", service);
-    Decider decider2 = replayDeciderCache.getOrCreate(decisionTask2, this::createFakeDecider);
+    Decider decider2 = replayDeciderCache.getOrCreate(decisionTask2, this::doNotCreateFakeDecider);
 
     // Assert
     // Wait for reporter
-    Thread.sleep(1000);
-    verify(reporter, times(1)).reportCounter(MetricsType.STICKY_CACHE_HIT, tags, 2);
-    assertEquals(decider2, replayDeciderCache.getUnchecked(runId));
-    assertEquals(decider2, decider);
+    Thread.sleep(500);
+    verify(reporter, times(1)).reportCounter(MetricsType.STICKY_CACHE_HIT, tags, 1);
+    assertEquals(decider, decider2);
+    service.close();
   }
 
   @Test
-  public void whenHistoryIsPartialAndCacheIsEmptyThenCacheEvictedExceptionIsThrown()
-      throws Exception {
+  public void whenHistoryIsPartialAndCacheIsEmptyThenExceptionIsThrown() throws Exception {
     // Arrange
     Map<String, String> tags =
         new ImmutableMap.Builder<String, String>(2)
@@ -139,7 +148,7 @@ public void whenHistoryIsPartialAndCacheIsEmptyThenCacheEvictedExceptionIsThrown
 
     try {
       replayDeciderCache.getOrCreate(decisionTask, this::createFakeDecider);
-    } catch (DeciderCache.EvictedException ex) {
+    } catch (IllegalArgumentException ex) {
 
       // Wait for reporter
       Thread.sleep(600);
@@ -148,7 +157,7 @@ public void whenHistoryIsPartialAndCacheIsEmptyThenCacheEvictedExceptionIsThrown
     }
 
     fail(
-        "Expected replayDeciderCache.getOrCreate to throw ReplayDeciderCache.EvictedException but no exception was thrown");
+        "Expected replayDeciderCache.getOrCreate to throw IllegalArgumentException but no exception was thrown");
   }
 
   @Test
@@ -208,15 +217,23 @@ public void evictAnyWillNotInvalidateItself() throws Exception {
   }
 
   private void assertCacheIsEmpty(DeciderCache cache, String runId) throws Exception {
-    DeciderCache.EvictedException ex = null;
+    Throwable ex = null;
     try {
-      cache.getUnchecked(runId);
-    } catch (DeciderCache.EvictedException e) {
+      PollForDecisionTaskResponse decisionTask =
+          new PollForDecisionTaskResponse()
+              .setWorkflowExecution(new WorkflowExecution().setRunId(runId));
+      cache.getOrCreate(decisionTask, this::doNotCreateFakeDecider);
+    } catch (AssertionError e) {
       ex = e;
     }
     TestCase.assertNotNull(ex);
   }
 
+  private ReplayDecider doNotCreateFakeDecider(PollForDecisionTaskResponse response) {
+    fail("should not be called");
+    return null;
+  }
+
   private ReplayDecider createFakeDecider(PollForDecisionTaskResponse response) {
     return new ReplayDecider(
         new TestWorkflowService(),