Fix activity_schedule_to_start_latency reporting activity_type unexpectedly

temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityPollTask.java

@@ -10,7 +10,7 @@
 import io.temporal.api.workflowservice.v1.GetSystemInfoResponse;
 import io.temporal.api.workflowservice.v1.PollActivityTaskQueueRequest;
 import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse;
-import io.temporal.internal.common.ProtobufTimeUtils;
+
 import io.temporal.serviceclient.MetricsTag;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.worker.MetricsType;
@@ -120,11 +120,6 @@ public ActivityTask poll() {
         metricsScope.counter(MetricsType.ACTIVITY_POLL_NO_TASK_COUNTER).inc(1);
         return null;
       }
-      metricsScope
-          .timer(MetricsType.ACTIVITY_SCHEDULE_TO_START_LATENCY)
-          .record(
-              ProtobufTimeUtils.toM3Duration(
-                  response.getStartedTime(), response.getCurrentAttemptScheduledTime()));
       isSuccessful = true;
       pollerTracker.pollSucceeded();
       return new ActivityTask(

temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncActivityPollTask.java

@@ -12,7 +12,7 @@
 import io.temporal.api.workflowservice.v1.PollActivityTaskQueueRequest;
 import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse;
 import io.temporal.internal.common.GrpcUtils;
-import io.temporal.internal.common.ProtobufTimeUtils;
+
 import io.temporal.serviceclient.MetricsTag;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.worker.MetricsType;
@@ -121,11 +121,6 @@ public CompletableFuture<ActivityTask> poll(SlotPermit permit) {
                 return null;
               }
               pollerTracker.pollSucceeded();
-              metricsScope
-                  .timer(MetricsType.ACTIVITY_SCHEDULE_TO_START_LATENCY)
-                  .record(
-                      ProtobufTimeUtils.toM3Duration(
-                          r.getStartedTime(), r.getCurrentAttemptScheduledTime()));
               return new ActivityTask(
                   r,
                   permit,

temporal-sdk/src/test/java/io/temporal/common/reporter/TestStatsReporter.java

@@ -89,6 +89,38 @@ public synchronized void assertTimer(String name, Map<String, String> tags) {
     }
   }
 
+  /**
+   * Asserts that a timer metric was reported exactly {@code expectedCount} times for the given
+   * tags. Useful for verifying that a metric is not accidentally recorded more than once (e.g.
+   * duplicated by multiple code paths).
+   */
+  public synchronized void assertTimer(String name, Map<String, String> tags, long expectedCount) {
+    String metricName = getMetricName(name, tags);
+    StatsAccumulator value = timers.get(metricName);
+    if (value == null) {
+      fail(
+          "No metric '"
+              + metricName
+              + "', reported metrics: \n "
+              + String.join("\n ", timers.keySet()));
+    }
+    assertEquals(
+        "Timer '" + metricName + "' was reported an unexpected number of times",
+        expectedCount,
+        value.count());
+  }
+
+  public synchronized void assertNoTimer(String name, Map<String, String> tags) {
+    String metricName = getMetricName(name, tags);
+    if (timers.containsKey(metricName)) {
+      fail(
+          "Timer metric '"
+              + metricName
+              + "' was reported, but was not expected. Recorded count: "
+              + timers.get(metricName).count());
+    }
+  }
+
   public synchronized void assertTimerMinDuration(
       String name, Map<String, String> tags, Duration minDuration) {
     String metricName = getMetricName(name, tags);

temporal-sdk/src/test/java/io/temporal/workflow/MetricsTest.java

@@ -120,6 +120,17 @@ public class MetricsTest {
           .put(MetricsTag.POLLER_TYPE, PollerTypeMetricsTag.PollerType.ACTIVITY_TASK.getValue())
           .build();
 
+  // Tags for the activity_schedule_to_start_latency metric, which is recorded with the real
+  // activity_type/workflow_type once per activity task in ActivityWorker.handleActivity (see
+  // #2733 - previously it was also incorrectly recorded a second/third time at poll time with
+  // activity_type defaulting to "none").
+  private static final Map<String, String> TAGS_ACTIVITY_SCHEDULE_TO_START =
+      new ImmutableMap.Builder<String, String>()
+          .putAll(TAGS_ACTIVITY_WORKER)
+          .put(MetricsTag.ACTIVITY_TYPE, "Execute")
+          .put(MetricsTag.WORKFLOW_TYPE, "NoArgsWorkflow")
+          .build();
+
   @Rule
   public TestWatcher watchman =
       new TestWatcher() {
@@ -183,7 +194,11 @@ public void testWorkerMetrics() throws InterruptedException {
         "temporal_workflow_task_queue_poll_succeed", TAGS_STICKY_WORKFLOW_WORKER);
     reporter.assertCounter("temporal_workflow_task_queue_poll_succeed", TAGS_WORKFLOW_WORKER);
     // We ran some workflow and activity tasks, so we should have some timers here.
-    reporter.assertTimer("temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_WORKER);
+    // activity_schedule_to_start_latency is recorded once per activity task, tagged with the
+    // real activity_type (see ActivityWorker.handleActivity / #2733), not the poller-level
+    // scope's default activity_type="none".
+    reporter.assertTimer(
+        "temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_SCHEDULE_TO_START);
     reporter.assertTimer("temporal_workflow_task_schedule_to_start_latency", TAGS_WORKFLOW_WORKER);
     reporter.assertTimer(
         "temporal_workflow_task_schedule_to_start_latency", TAGS_STICKY_WORKFLOW_WORKER);
@@ -228,7 +243,11 @@ public void testWorkerMetricsAutoPoller() throws InterruptedException {
         "temporal_workflow_task_queue_poll_succeed", TAGS_STICKY_WORKFLOW_WORKER);
     reporter.assertCounter("temporal_workflow_task_queue_poll_succeed", TAGS_WORKFLOW_WORKER);
     // We ran some workflow and activity tasks, so we should have some timers here.
-    reporter.assertTimer("temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_WORKER);
+    // activity_schedule_to_start_latency is recorded once per activity task, tagged with the
+    // real activity_type (see ActivityWorker.handleActivity / #2733), not the poller-level
+    // scope's default activity_type="none".
+    reporter.assertTimer(
+        "temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_SCHEDULE_TO_START);
     reporter.assertTimer("temporal_workflow_task_schedule_to_start_latency", TAGS_WORKFLOW_WORKER);
     reporter.assertTimer(
         "temporal_workflow_task_schedule_to_start_latency", TAGS_STICKY_WORKFLOW_WORKER);
@@ -239,6 +258,47 @@ public void testWorkerMetricsAutoPoller() throws InterruptedException {
     reporter.assertGauge("temporal_num_pollers", TAGS_ACTIVITY_POLLER, 5);
   }
 
+  /**
+   * Regression test for #2733: activity_schedule_to_start_latency must be recorded exactly once per
+   * activity task, and only with the correct activity_type tag.
+   *
+   * <p>Previously the metric was recorded up to three times per activity task: once in {@code
+   * ActivityPollTask}/{@code AsyncActivityPollTask} at poll time (using the poller-level scope,
+   * where activity_type defaults to "none"), and once in {@code ActivityWorker.handleActivity}
+   * using a scope tagged with the real activity_type. This test verifies the duplicate poll-time
+   * recording(s) were removed and only the correctly-tagged recording from {@code handleActivity}
+   * remains.
+   */
+  @Test
+  public void testActivityScheduleToStartLatencyMetricRecordedOnce() throws InterruptedException {
+    setUp(WorkerFactoryOptions.getDefaultInstance());
+
+    Worker worker = testEnvironment.newWorker(TASK_QUEUE);
+    worker.registerWorkflowImplementationTypes(TestScheduleToStartMetricWorkflow.class);
+    worker.registerActivitiesImplementations(new TestActivityImpl());
+    testEnvironment.start();
+
+    WorkflowClient workflowClient = testEnvironment.getWorkflowClient();
+    WorkflowOptions options =
+        WorkflowOptions.newBuilder()
+            .setWorkflowRunTimeout(Duration.ofSeconds(1000))
+            .setTaskQueue(TASK_QUEUE)
+            .build();
+    NoArgsWorkflow workflow = workflowClient.newWorkflowStub(NoArgsWorkflow.class, options);
+    workflow.execute();
+
+    Thread.sleep(REPORTING_FLUSH_TIME);
+
+    // The metric must be recorded exactly once with the correct activity_type/workflow_type
+    // tags (i.e. the recording made by ActivityWorker.handleActivity).
+    reporter.assertTimer(
+        "temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_SCHEDULE_TO_START, 1);
+
+    // The metric must NOT additionally be recorded under the poller-level scope, where
+    // activity_type defaults to "none" - that was the duplicate recording removed by #2733.
+    reporter.assertNoTimer("temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_WORKER);
+  }
+
   @Test
   public void testWorkflowMetrics() throws InterruptedException {
     setUp(WorkerFactoryOptions.getDefaultInstance());
@@ -593,6 +653,20 @@ public int execute(int input) {
     }
   }
 
+  public static class TestScheduleToStartMetricWorkflow implements NoArgsWorkflow {
+
+    @Override
+    public void execute() {
+      ActivityOptions activityOptions =
+          ActivityOptions.newBuilder()
+              .setTaskQueue(TASK_QUEUE)
+              .setScheduleToCloseTimeout(Duration.ofSeconds(100))
+              .build();
+      TestActivity3 activity = Workflow.newActivityStub(TestActivity3.class, activityOptions);
+      activity.execute(1);
+    }
+  }
+
   public static class TestMetricsInChildWorkflow implements TestChildWorkflow {
 
     @Override