From 6cb2c04b4efa90bff85257e4ddcb9599b301ab4f Mon Sep 17 00:00:00 2001 From: sachinsharma Date: Sun, 9 Aug 2026 18:17:17 -0700 Subject: [PATCH 1/2] Fix activity_schedule_to_start_latency reporting activity_type="none" The metric was recorded in three places: twice at poll time (sync and async pollers) with the default activity_type="none", and once in ActivityWorker.handleActivity with the correct activity_type tag. Remove the duplicate poll-time recordings so the metric is emitted exactly once per activity task with the proper activity_type. Fixes #2733 --- .../java/io/temporal/internal/worker/ActivityPollTask.java | 7 +------ .../io/temporal/internal/worker/AsyncActivityPollTask.java | 7 +------ 2 files changed, 2 insertions(+), 12 deletions(-) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityPollTask.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityPollTask.java index f0d3e649f..c92cdb4ec 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityPollTask.java +++ b/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( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncActivityPollTask.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncActivityPollTask.java index 1e8791bd0..f8766c110 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncActivityPollTask.java +++ b/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 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, From 40514406d3d2ac7db4342965786e15bdc4a827b1 Mon Sep 17 00:00:00 2001 From: sachinsharma Date: Mon, 17 Aug 2026 16:23:50 -0700 Subject: [PATCH 2/2] Add regression tests for activity_schedule_to_start_latency duplicate fix Adds testActivityScheduleToStartLatencyMetricRecordedOnce to MetricsTest, which verifies the metric is recorded exactly once per activity task with the correct activity_type/workflow_type tags, and is not additionally recorded under the poller-level scope where activity_type defaults to "none" (the duplicate recording removed in this PR, see #2733). Also fixes the now-stale assertions in testWorkerMetrics and testWorkerMetricsAutoPoller, which asserted the metric under the untagged/"none" poller scope and would otherwise fail now that the duplicate poll-time recording has been removed. TestStatsReporter gains assertTimer(name, tags, expectedCount) and assertNoTimer(name, tags) helpers to support asserting exact recording counts and absence of a timer metric. --- .../common/reporter/TestStatsReporter.java | 32 ++++++++ .../io/temporal/workflow/MetricsTest.java | 78 ++++++++++++++++++- 2 files changed, 108 insertions(+), 2 deletions(-) diff --git a/temporal-sdk/src/test/java/io/temporal/common/reporter/TestStatsReporter.java b/temporal-sdk/src/test/java/io/temporal/common/reporter/TestStatsReporter.java index 6e0327816..00a2f47fd 100644 --- a/temporal-sdk/src/test/java/io/temporal/common/reporter/TestStatsReporter.java +++ b/temporal-sdk/src/test/java/io/temporal/common/reporter/TestStatsReporter.java @@ -89,6 +89,38 @@ public synchronized void assertTimer(String name, Map 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 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 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 tags, Duration minDuration) { String metricName = getMetricName(name, tags); diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/MetricsTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/MetricsTest.java index 17178b199..4cc1de332 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/MetricsTest.java +++ b/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 TAGS_ACTIVITY_SCHEDULE_TO_START = + new ImmutableMap.Builder() + .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. + * + *

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