Add metrics tags and set correct data converter in test worker creation (#170)

* update gradle build script to prepare for publishing to maven central

* do not sign archives if key is missing

* Add domain and task list metric tags, use correct data converter in test worker creation

Author: meiliang86

build.gradle

@@ -37,7 +37,7 @@ googleJavaFormat {
 }
 
 group = 'com.uber.cadence'
-version = '0.2.0-SNAPSHOT'
+version = '0.2.1'
 
 description = """Uber Cadence Java Client"""
 

src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java

@@ -110,16 +110,16 @@ public TestWorkflowEnvironmentInternal(TestEnvironmentOptions options) {
 
   @Override
   public Worker newWorker(String taskList) {
+    WorkerOptions.Builder builder =
+        new WorkerOptions.Builder()
+            .setInterceptorFactory(testEnvironmentOptions.getInterceptorFactory())
+            .setMetricsScope(testEnvironmentOptions.getMetricsScope())
+            .setEnableLoggingInReplay(testEnvironmentOptions.isLoggingEnabledInReplay());
+    if (testEnvironmentOptions.getDataConverter() != null) {
+      builder.setDataConverter(testEnvironmentOptions.getDataConverter());
+    }
     Worker result =
-        new Worker(
-            service,
-            testEnvironmentOptions.getDomain(),
-            taskList,
-            new WorkerOptions.Builder()
-                .setInterceptorFactory(testEnvironmentOptions.getInterceptorFactory())
-                .setMetricsScope(testEnvironmentOptions.getMetricsScope())
-                .setEnableLoggingInReplay(testEnvironmentOptions.isLoggingEnabledInReplay())
-                .build());
+        new Worker(service, testEnvironmentOptions.getDomain(), taskList, builder.build());
     workers.add(result);
     return result;
   }

src/main/java/com/uber/cadence/worker/Worker.java

@@ -21,6 +21,7 @@
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.client.WorkflowClient;
 import com.uber.cadence.converter.DataConverter;
+import com.uber.cadence.internal.metrics.MetricsTag;
 import com.uber.cadence.internal.sync.SyncActivityWorker;
 import com.uber.cadence.internal.sync.SyncWorkflowWorker;
 import com.uber.cadence.internal.worker.SingleWorkerOptions;
@@ -29,7 +30,9 @@
 import com.uber.cadence.worker.WorkerOptions.Builder;
 import com.uber.cadence.workflow.Functions.Func;
 import com.uber.cadence.workflow.WorkflowMethod;
+import com.uber.m3.util.ImmutableMap;
 import java.time.Duration;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -115,13 +118,13 @@ public Worker(IWorkflowService service, String domain, String taskList, WorkerOp
       options = new Builder().build();
     }
     this.options = options;
-    SingleWorkerOptions activityOptions = toActivityOptions(options);
+    SingleWorkerOptions activityOptions = toActivityOptions(options, domain, taskList);
     if (!options.isDisableActivityWorker()) {
       activityWorker = new SyncActivityWorker(service, domain, taskList, activityOptions);
     } else {
       activityWorker = null;
     }
-    SingleWorkerOptions workflowOptions = toWorkflowOptions(options);
+    SingleWorkerOptions workflowOptions = toWorkflowOptions(options, domain, taskList);
     if (!options.isDisableWorkflowWorker()) {
       workflowWorker =
           new SyncWorkflowWorker(
@@ -136,28 +139,40 @@ public Worker(IWorkflowService service, String domain, String taskList, WorkerOp
     }
   }
 
-  private SingleWorkerOptions toActivityOptions(WorkerOptions options) {
+  private SingleWorkerOptions toActivityOptions(
+      WorkerOptions options, String domain, String taskList) {
+    Map<String, String> tags =
+        new ImmutableMap.Builder<String, String>(2)
+            .put(MetricsTag.DOMAIN, domain)
+            .put(MetricsTag.TASK_LIST, taskList)
+            .build();
     return new SingleWorkerOptions.Builder()
         .setDataConverter(options.getDataConverter())
         .setIdentity(options.getIdentity())
         .setPollerOptions(options.getActivityPollerOptions())
         .setReportCompletionRetryOptions(options.getReportActivityCompletionRetryOptions())
         .setReportFailureRetryOptions(options.getReportActivityFailureRetryOptions())
         .setTaskExecutorThreadPoolSize(options.getMaxConcurrentActivityExecutionSize())
-        .setMetricsScope(options.getMetricsScope())
+        .setMetricsScope(options.getMetricsScope().tagged(tags))
         .setEnableLoggingInReplay(options.getEnableLoggingInReplay())
         .build();
   }
 
-  private SingleWorkerOptions toWorkflowOptions(WorkerOptions options) {
+  private SingleWorkerOptions toWorkflowOptions(
+      WorkerOptions options, String domain, String taskList) {
+    Map<String, String> tags =
+        new ImmutableMap.Builder<String, String>(2)
+            .put(MetricsTag.DOMAIN, domain)
+            .put(MetricsTag.TASK_LIST, taskList)
+            .build();
     return new SingleWorkerOptions.Builder()
         .setDataConverter(options.getDataConverter())
         .setIdentity(options.getIdentity())
         .setPollerOptions(options.getWorkflowPollerOptions())
         .setReportCompletionRetryOptions(options.getReportWorkflowCompletionRetryOptions())
         .setReportFailureRetryOptions(options.getReportWorkflowFailureRetryOptions())
         .setTaskExecutorThreadPoolSize(options.getMaxConcurrentWorklfowExecutionSize())
-        .setMetricsScope(options.getMetricsScope())
+        .setMetricsScope(options.getMetricsScope().tagged(tags))
         .setEnableLoggingInReplay(options.getEnableLoggingInReplay())
         .build();
   }

src/test/java/com/uber/cadence/workflow/MetricsTest.java

@@ -26,6 +26,7 @@
 
 import com.uber.cadence.client.WorkflowClient;
 import com.uber.cadence.client.WorkflowOptions;
+import com.uber.cadence.internal.metrics.MetricsTag;
 import com.uber.cadence.testing.TestEnvironmentOptions;
 import com.uber.cadence.testing.TestEnvironmentOptions.Builder;
 import com.uber.cadence.testing.TestWorkflowEnvironment;
@@ -34,7 +35,9 @@
 import com.uber.m3.tally.Scope;
 import com.uber.m3.tally.StatsReporter;
 import com.uber.m3.tally.Stopwatch;
+import com.uber.m3.util.ImmutableMap;
 import java.time.Duration;
+import java.util.Map;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -132,10 +135,16 @@ public void testWorkflowMetrics() throws InterruptedException {
 
     Thread.sleep(200);
 
-    verify(reporter, times(1)).reportCounter("test-started", null, 1);
-    verify(reporter, times(1)).reportCounter("test-done", null, 1);
-    verify(reporter, times(1)).reportCounter("test-child-started", null, 1);
-    verify(reporter, times(1)).reportCounter("test-child-done", null, 1);
+    Map<String, String> tags =
+        new ImmutableMap.Builder<String, String>(2)
+            .put(MetricsTag.DOMAIN, WorkflowTest.DOMAIN)
+            .put(MetricsTag.TASK_LIST, taskList)
+            .build();
+
+    verify(reporter, times(1)).reportCounter("test-started", tags, 1);
+    verify(reporter, times(1)).reportCounter("test-done", tags, 1);
+    verify(reporter, times(1)).reportCounter("test-child-started", tags, 1);
+    verify(reporter, times(1)).reportCounter("test-child-done", tags, 1);
 
     ArgumentCaptor<com.uber.m3.util.Duration> sleepDurationCaptor =
         ArgumentCaptor.forClass(com.uber.m3.util.Duration.class);