Report activity schedule to start latency once per task without the activity type

[RaphaelFakhri]

What was changed

• Removed the activity_schedule_to_start_latency timer from ActivityPollTask and AsyncActivityPollTask.
• Moved the remaining recording from ActivityWorker.TaskHandlerImpl.handleActivity into TaskHandlerImpl.handle, just before the call to handleActivity. It now uses workerMetricsScope, so the metric carries no activity type or workflow type tags.
• Added a TestStatsReporter.assertTimer overload that takes a Predicate<StatsAccumulator>, modelled on assertGauge.
• TestStatsReporter.assertNoMetric now also checks timers. It only checked counters before.
• MetricsTest.testWorkerMetrics asserts that the timer is reported exactly once on the activity worker tags and never on the tags that include the activity type.

Why?

Each activity task recorded the schedule to start latency twice: once in the poll task and once in handleActivity with the activity type tags. Schedule to start latency does not depend on the activity type, so it is now recorded once per task on the worker scope.

How was this tested?

./gradlew :temporal-sdk:test --tests io.temporal.workflow.MetricsTest

With the change, all 9 tests pass. With the three source files reverted to main, testWorkerMetrics fails because the timer is reported with activity_type=Execute and workflow_type=NoArgsWorkflow.

Closes #2733

[CLAassistant]

All committers have signed the CLA.

[maciejdudko]

Hi @RaphaelFakhri, thank you for your contribution! However, this isn't the full fix for the issue.

Emitting the metric was correctly removed from ActivityPollTask and AsyncActivityPollTask. However, emitting it inside ActivityWorker.TaskHandlerImpl.handleActivity is still wrong. It should be moved into TaskHandlerImpl.handle method, just before the call to handleActivity. It also should use workerMetricsScope instead of the local metricsScope, because attaching worker and activity type to this metric is wrong (schedule-to-start latency is independent of activity type).

I left a separate comment about tests, please address it too.

Feel free to reach out to me on Temporal Community Slack if you need assistance with implementation.

[maciejdudko]

This new test suite is unnecessary. It's better to amend the existing io.temporal.workflow.MetricsTest. To ensure the metric is emitted only once, you can add a new overload of TestStatsReporter.assertTimer that takes Predicate<StatsAccumulator>, similar to how assertGauge works. To ensure the metric is not sent with wrong set of tags, you can use assertNoMetric.

[RaphaelFakhri]

Thanks for the review. The timer is now recorded once in TaskHandlerImpl.handle on workerMetricsScope, and the assertions live in MetricsTest with the new assertTimer overload.