Skip to content

Commit dcb6b69

Browse files
committed
Report poller target in the worker heartbeat
The poller target is already sent on each poll, but matching consumes it inside the scaling decision and discards it -- nothing persists it and nothing can observe it. Reporting it on the heartbeat too lets the server read it per worker and export it as a metric, making fleet-wide poller distribution visible without instrumenting every SDK. WorkerHeartbeat already carries four WorkerPollerInfo sub-messages, so this is per-task-type for free, including the normal-vs-sticky workflow split. PollScaleReportHandle is a local inside AsyncPoller.start() and is not reachable from the heartbeat builder. PollerTracker already exists to bridge exactly that gap -- its javadoc says so -- so the target supplier goes there, and buildPollerInfo reads it alongside the in-flight count it already reads. AsyncNexusPollTask now overrides setPollerTargetSupplier purely to feed its tracker. It had taken the interface's no-op default because the nexus poll request has no field for a target, but it does have an autoscaler and a heartbeat slot. Fixed-size pools run MultiThreadedPoller, which has no scale report handle, so no supplier is installed and the tracker reports 0 -- which the proto defines as unknown. The api submodule moves to the commit carrying the new field.
1 parent 83ec097 commit dcb6b69

6 files changed

Lines changed: 37 additions & 1 deletion

File tree

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,8 @@ public AsyncActivityPollTask(
9797
@Override
9898
public void setPollerTargetSupplier(IntSupplier pollerTargetSupplier) {
9999
this.pollerTargetSupplier = pollerTargetSupplier;
100+
// Also hand it to the tracker, which is what the heartbeat builder can reach.
101+
pollerTracker.setPollerTargetSupplier(pollerTargetSupplier);
100102
}
101103

102104
/**

‎temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncNexusPollTask.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import io.temporal.worker.tuning.SlotReleaseReason;
2222
import java.util.Objects;
2323
import java.util.concurrent.CompletableFuture;
24+
import java.util.function.IntSupplier;
2425
import java.util.function.Supplier;
2526
import javax.annotation.Nonnull;
2627
import org.slf4j.Logger;
@@ -110,6 +111,16 @@ public AsyncNexusPollTask(
110111
this.pollRequest = pollRequest.build();
111112
}
112113

114+
/**
115+
* The nexus poll request has no field for a poller target, so unlike the workflow and activity
116+
* tasks this only feeds the tracker -- the heartbeat reports a target for every autoscaling pool,
117+
* not only the pools poll-time fairness acts on.
118+
*/
119+
@Override
120+
public void setPollerTargetSupplier(IntSupplier pollerTargetSupplier) {
121+
pollerTracker.setPollerTargetSupplier(pollerTargetSupplier);
122+
}
123+
113124
@Override
114125
@SuppressWarnings("deprecation")
115126
public CompletableFuture<NexusTask> poll(SlotPermit permit) {

‎temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncWorkflowPollTask.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,8 @@ public AsyncWorkflowPollTask(
129129
@Override
130130
public void setPollerTargetSupplier(IntSupplier pollerTargetSupplier) {
131131
this.pollerTargetSupplier = pollerTargetSupplier;
132+
// Also hand it to the tracker, which is what the heartbeat builder can reach.
133+
pollerTracker.setPollerTargetSupplier(pollerTargetSupplier);
132134
}
133135

134136
/**

‎temporal-sdk/src/main/java/io/temporal/internal/worker/PollerTracker.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import java.time.Instant;
44
import java.util.concurrent.atomic.AtomicInteger;
55
import java.util.concurrent.atomic.AtomicReference;
6+
import java.util.function.IntSupplier;
67

78
/**
89
* Tracks in-flight poll count and last successful poll time for heartbeat reporting.
@@ -20,6 +21,7 @@
2021
public class PollerTracker {
2122
private final AtomicInteger inFlightPolls = new AtomicInteger();
2223
private final AtomicReference<Instant> lastSuccessfulPollTime = new AtomicReference<>();
24+
private final AtomicReference<IntSupplier> pollerTargetSupplier = new AtomicReference<>();
2325

2426
/** Increments in-flight count. Returns the new value for forwarding to the Tally gauge. */
2527
public int pollStarted() {
@@ -43,4 +45,22 @@ public int getInFlightPolls() {
4345
public Instant getLastSuccessfulPollTime() {
4446
return lastSuccessfulPollTime.get();
4547
}
48+
49+
/**
50+
* Supplies the autoscaler's current target for this pool. Set by the poll task once {@link
51+
* AsyncPoller} has installed it, which is why this is a supplier rather than a value -- the scale
52+
* report handle does not exist until the poller starts.
53+
*/
54+
public void setPollerTargetSupplier(IntSupplier supplier) {
55+
pollerTargetSupplier.set(supplier);
56+
}
57+
58+
/**
59+
* The autoscaler's current target, or 0 when this pool has a fixed size and so has no target. The
60+
* heartbeat proto defines 0 as unknown.
61+
*/
62+
public int getPollerTarget() {
63+
IntSupplier supplier = pollerTargetSupplier.get();
64+
return supplier == null ? 0 : supplier.getAsInt();
65+
}
4666
}

‎temporal-sdk/src/main/java/io/temporal/worker/Worker.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -749,6 +749,7 @@ private static WorkerPollerInfo buildPollerInfo(
749749
builder.setIsAutoscaling(true);
750750
}
751751
builder.setCurrentPollers(tracker.getInFlightPolls());
752+
builder.setPollerTarget(tracker.getPollerTarget());
752753
Instant lastPoll = tracker.getLastSuccessfulPollTime();
753754
if (lastPoll != null) {
754755
builder.setLastSuccessfulPollTime(toProtoTimestamp(lastPoll));

0 commit comments

Comments
 (0)