#348

Committed 31 Jul 2026 10:02PM UTC coverage: 68.143% (-0.005%) from 68.148%

Build # #348

Build Type

push

github

Committed by

web-flow

Commit Message

Do not send versioning info on worker command channel polls (#2987)

Coverage Stats

7166 of 12558 branches covered (57.06%)

Branch coverage included in aggregate %.

6 of 22 new or added lines in 2 files covered. (27.27%)

2 existing lines in 2 files now uncovered.

29723 of 41577 relevant lines covered (71.49%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.32

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

package io.temporal.internal.worker;

import static io.temporal.serviceclient.MetricsTag.METRICS_TAGS_CALL_OPTIONS_KEY;

import com.google.protobuf.Timestamp;
import com.uber.m3.tally.Scope;
import io.grpc.Context;
import io.temporal.api.common.v1.WorkerVersionCapabilities;
import io.temporal.api.enums.v1.TaskQueueKind;
import io.temporal.api.taskqueue.v1.TaskQueue;
import io.temporal.api.workflowservice.v1.GetSystemInfoResponse;
import io.temporal.api.workflowservice.v1.PollNexusTaskQueueRequest;
import io.temporal.api.workflowservice.v1.PollNexusTaskQueueResponse;
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;
import io.temporal.worker.PollerTypeMetricsTag;
import io.temporal.worker.tuning.SlotPermit;
import io.temporal.worker.tuning.SlotReleaseReason;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import javax.annotation.Nonnull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class AsyncNexusPollTask implements AsyncPoller.PollTaskAsync<NexusTask> {
  private static final Logger log = LoggerFactory.getLogger(AsyncNexusPollTask.class);

  private final TrackingSlotSupplier<?> slotSupplier;
  private final WorkflowServiceStubs service;
  private final Scope metricsScope;
  private final PollNexusTaskQueueRequest pollRequest;
  private final Context.CancellableContext grpcContext = Context.ROOT.withCancellation();
  private final PollerTracker pollerTracker;

  @SuppressWarnings("deprecation")
  public AsyncNexusPollTask(
      @Nonnull WorkflowServiceStubs service,
      @Nonnull String namespace,
      @Nonnull String taskQueue,
      @Nonnull String identity,
      @Nonnull String workerInstanceKey,
      @Nonnull WorkerVersioningOptions versioningOptions,
      @Nonnull Scope metricsScope,
      @Nonnull Supplier<GetSystemInfoResponse.Capabilities> serverCapabilities,
      TrackingSlotSupplier<?> slotSupplier,
      @Nonnull PollerTracker pollerTracker) {
    this(
        service,
        namespace,
        taskQueue,
        identity,
        workerInstanceKey,
        versioningOptions,
        metricsScope,
        serverCapabilities,
        slotSupplier,
        pollerTracker,
        false);
  }

  @SuppressWarnings("deprecation")
  public AsyncNexusPollTask(
      @Nonnull WorkflowServiceStubs service,
      @Nonnull String namespace,
      @Nonnull String taskQueue,
      @Nonnull String identity,
      @Nonnull String workerInstanceKey,
      @Nonnull WorkerVersioningOptions versioningOptions,
      @Nonnull Scope metricsScope,
      @Nonnull Supplier<GetSystemInfoResponse.Capabilities> serverCapabilities,
      TrackingSlotSupplier<?> slotSupplier,
      @Nonnull PollerTracker pollerTracker,
      boolean workerCommandsTaskQueue) {
    this.service = Objects.requireNonNull(service);
    this.metricsScope = Objects.requireNonNull(metricsScope);
    this.slotSupplier = slotSupplier;
    this.pollerTracker = Objects.requireNonNull(pollerTracker);

    PollNexusTaskQueueRequest.Builder pollRequest =
        PollNexusTaskQueueRequest.newBuilder()
            .setNamespace(namespace)
            .setIdentity(identity)
            .setTaskQueue(
                TaskQueue.newBuilder()
                    .setName(taskQueue)
                    .setKind(
                        workerCommandsTaskQueue
                            ? TaskQueueKind.TASK_QUEUE_KIND_WORKER_COMMANDS
                            : TaskQueueKind.TASK_QUEUE_KIND_NORMAL));

    pollRequest.setWorkerInstanceKey(workerInstanceKey);

    if (!workerCommandsTaskQueue) {
      if (versioningOptions.getWorkerDeploymentOptions() != null) {
        pollRequest.setDeploymentOptions(
            WorkerVersioningProtoUtils.deploymentOptionsToProto(
                versioningOptions.getWorkerDeploymentOptions()));
      } else if (serverCapabilities.get().getBuildIdBasedVersioning()) {
        pollRequest.setWorkerVersionCapabilities(
            WorkerVersionCapabilities.newBuilder()
                .setBuildId(versioningOptions.getBuildId())
                .setUseVersioning(versioningOptions.isUsingVersioning())
                .build());
      }
    }
    this.pollRequest = pollRequest.build();
  }

  @Override
  @SuppressWarnings("deprecation")
  public CompletableFuture<NexusTask> poll(SlotPermit permit) {
    if (log.isTraceEnabled()) {
      log.trace("poll request begin: " + pollRequest);
    }

    MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.NEXUS_TASK)
        .gauge(MetricsType.NUM_POLLERS)
        .update(pollerTracker.pollStarted());

    CompletableFuture<PollNexusTaskQueueResponse> response = null;
    try {
      response =
          grpcContext.call(
              () ->
                  GrpcUtils.toCompletableFuture(
                      service
                          .futureStub()
                          .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope)
                          .pollNexusTaskQueue(pollRequest)));
    } catch (Exception e) {
      MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.NEXUS_TASK)
          .gauge(MetricsType.NUM_POLLERS)
          .update(pollerTracker.pollCompleted());
      throw new RuntimeException(e);
    }

    return response
        .thenApply(
            r -> {
              if (r == null || r.getTaskToken().isEmpty()) {
                metricsScope.counter(MetricsType.NEXUS_POLL_NO_TASK_COUNTER).inc(1);
                return null;
              }
              pollerTracker.pollSucceeded();
              Timestamp startedTime = ProtobufTimeUtils.getCurrentProtoTime();
              metricsScope
                  .timer(MetricsType.NEXUS_SCHEDULE_TO_START_LATENCY)
                  .record(
                      ProtobufTimeUtils.toM3Duration(
                          startedTime, r.getRequest().getScheduledTime()));
              return new NexusTask(
                  r,
                  permit,
                  () -> slotSupplier.releaseSlot(SlotReleaseReason.taskComplete(), permit));
            })
        .whenComplete(
            (r, e) -> {
              MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.NEXUS_TASK)
                  .gauge(MetricsType.NUM_POLLERS)
                  .update(pollerTracker.pollCompleted());
            });
  }

  @Override
  public void cancel(Throwable cause) {
    grpcContext.cancel(cause);
  }

  @Override
  public String getLabel() {
    return "AsyncNexusPollTask";
  }

  @Override
  public String toString() {
    return "AsyncNexusPollTask{}";
  }
}

1	package io.temporal.internal.worker;
2
3	import static io.temporal.serviceclient.MetricsTag.METRICS_TAGS_CALL_OPTIONS_KEY;
4
5	import com.google.protobuf.Timestamp;
6	import com.uber.m3.tally.Scope;
7	import io.grpc.Context;
8	import io.temporal.api.common.v1.WorkerVersionCapabilities;
9	import io.temporal.api.enums.v1.TaskQueueKind;
10	import io.temporal.api.taskqueue.v1.TaskQueue;
11	import io.temporal.api.workflowservice.v1.GetSystemInfoResponse;
12	import io.temporal.api.workflowservice.v1.PollNexusTaskQueueRequest;
13	import io.temporal.api.workflowservice.v1.PollNexusTaskQueueResponse;
14	import io.temporal.internal.common.GrpcUtils;
15	import io.temporal.internal.common.ProtobufTimeUtils;
16	import io.temporal.serviceclient.MetricsTag;
17	import io.temporal.serviceclient.WorkflowServiceStubs;
18	import io.temporal.worker.MetricsType;
19	import io.temporal.worker.PollerTypeMetricsTag;
20	import io.temporal.worker.tuning.SlotPermit;
21	import io.temporal.worker.tuning.SlotReleaseReason;
22	import java.util.Objects;
23	import java.util.concurrent.CompletableFuture;
24	import java.util.function.Supplier;
25	import javax.annotation.Nonnull;
26	import org.slf4j.Logger;
27	import org.slf4j.LoggerFactory;
28
29	public class AsyncNexusPollTask implements AsyncPoller.PollTaskAsync<NexusTask> {
30	private static final Logger log = LoggerFactory.getLogger(AsyncNexusPollTask.class);	1✔
31
32	private final TrackingSlotSupplier<?> slotSupplier;
33	private final WorkflowServiceStubs service;
34	private final Scope metricsScope;
35	private final PollNexusTaskQueueRequest pollRequest;
36	private final Context.CancellableContext grpcContext = Context.ROOT.withCancellation();	1✔
37	private final PollerTracker pollerTracker;
38
39	@SuppressWarnings("deprecation")
40	public AsyncNexusPollTask(
41	@Nonnull WorkflowServiceStubs service,
42	@Nonnull String namespace,
43	@Nonnull String taskQueue,
44	@Nonnull String identity,
45	@Nonnull String workerInstanceKey,
46	@Nonnull WorkerVersioningOptions versioningOptions,
47	@Nonnull Scope metricsScope,
48	@Nonnull Supplier<GetSystemInfoResponse.Capabilities> serverCapabilities,
49	TrackingSlotSupplier<?> slotSupplier,
50	@Nonnull PollerTracker pollerTracker) {
51	this(	×
52	service,
53	namespace,
54	taskQueue,
55	identity,
56	workerInstanceKey,
57	versioningOptions,
58	metricsScope,
59	serverCapabilities,
60	slotSupplier,
61	pollerTracker,
62	false);
63	}	×
64
65	@SuppressWarnings("deprecation")
66	public AsyncNexusPollTask(
67	@Nonnull WorkflowServiceStubs service,
68	@Nonnull String namespace,
69	@Nonnull String taskQueue,
70	@Nonnull String identity,
71	@Nonnull String workerInstanceKey,
72	@Nonnull WorkerVersioningOptions versioningOptions,
73	@Nonnull Scope metricsScope,
74	@Nonnull Supplier<GetSystemInfoResponse.Capabilities> serverCapabilities,
75	TrackingSlotSupplier<?> slotSupplier,
76	@Nonnull PollerTracker pollerTracker,
77	boolean workerCommandsTaskQueue) {	1✔
78	this.service = Objects.requireNonNull(service);	1✔
79	this.metricsScope = Objects.requireNonNull(metricsScope);	1✔
80	this.slotSupplier = slotSupplier;	1✔
81	this.pollerTracker = Objects.requireNonNull(pollerTracker);	1✔
82
83	PollNexusTaskQueueRequest.Builder pollRequest =
84	PollNexusTaskQueueRequest.newBuilder()	1✔
85	.setNamespace(namespace)	1✔
86	.setIdentity(identity)	1✔
87	.setTaskQueue(	1✔
88	TaskQueue.newBuilder()	1✔
89	.setName(taskQueue)	1✔
90	.setKind(	1✔
91	workerCommandsTaskQueue	1!
92	? TaskQueueKind.TASK_QUEUE_KIND_WORKER_COMMANDS	×
93	: TaskQueueKind.TASK_QUEUE_KIND_NORMAL));	1✔
94
95	pollRequest.setWorkerInstanceKey(workerInstanceKey);	1✔
96
97	if (!workerCommandsTaskQueue) {	1!
98	if (versioningOptions.getWorkerDeploymentOptions() != null) {	1!
NEW 99	pollRequest.setDeploymentOptions(	×
NEW 100	WorkerVersioningProtoUtils.deploymentOptionsToProto(	×
NEW 101	versioningOptions.getWorkerDeploymentOptions()));	×
102	} else if (serverCapabilities.get().getBuildIdBasedVersioning()) {	1!
NEW 103	pollRequest.setWorkerVersionCapabilities(	×
NEW 104	WorkerVersionCapabilities.newBuilder()	×
NEW 105	.setBuildId(versioningOptions.getBuildId())	×
NEW 106	.setUseVersioning(versioningOptions.isUsingVersioning())	×
NEW 107	.build());	×
108	}
109	}
110	this.pollRequest = pollRequest.build();	1✔
111	}	1✔
112
113	@Override
114	@SuppressWarnings("deprecation")
115	public CompletableFuture<NexusTask> poll(SlotPermit permit) {
116	if (log.isTraceEnabled()) {	1!
117	log.trace("poll request begin: " + pollRequest);	×
118	}
119
120	MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.NEXUS_TASK)	1✔
121	.gauge(MetricsType.NUM_POLLERS)	1✔
122	.update(pollerTracker.pollStarted());	1✔
123
124	CompletableFuture<PollNexusTaskQueueResponse> response = null;	1✔
125	try {
126	response =	1✔
127	grpcContext.call(	1✔
128	() ->
129	GrpcUtils.toCompletableFuture(	1✔
130	service
131	.futureStub()	1✔
132	.withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope)	1✔
133	.pollNexusTaskQueue(pollRequest)));	1✔
134	} catch (Exception e) {	×
135	MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.NEXUS_TASK)	×
136	.gauge(MetricsType.NUM_POLLERS)	×
137	.update(pollerTracker.pollCompleted());	×
138	throw new RuntimeException(e);	×
139	}	1✔
140
141	return response	1✔
142	.thenApply(	1✔
143	r -> {
144	if (r == null \|\| r.getTaskToken().isEmpty()) {	1!
145	metricsScope.counter(MetricsType.NEXUS_POLL_NO_TASK_COUNTER).inc(1);	×
146	return null;	×
147	}
148	pollerTracker.pollSucceeded();	1✔
149	Timestamp startedTime = ProtobufTimeUtils.getCurrentProtoTime();	1✔
150	metricsScope	1✔
151	.timer(MetricsType.NEXUS_SCHEDULE_TO_START_LATENCY)	1✔
152	.record(	1✔
153	ProtobufTimeUtils.toM3Duration(	1✔
154	startedTime, r.getRequest().getScheduledTime()));	1✔
155	return new NexusTask(	1✔
156	r,
157	permit,
158	() -> slotSupplier.releaseSlot(SlotReleaseReason.taskComplete(), permit));	1✔
159	})
160	.whenComplete(	1✔
161	(r, e) -> {
162	MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.NEXUS_TASK)	1✔
163	.gauge(MetricsType.NUM_POLLERS)	1✔
164	.update(pollerTracker.pollCompleted());	1✔
165	});	1✔
166	}
167
168	@Override
169	public void cancel(Throwable cause) {
170	grpcContext.cancel(cause);	1✔
171	}	1✔
172
173	@Override
174	public String getLabel() {
175	return "AsyncNexusPollTask";	1✔
176	}
177
178	@Override
179	public String toString() {
180	return "AsyncNexusPollTask{}";	×
181	}
182	}

temporalio / sdk-java / #348

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous