#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

75.0

/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusPollTask.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.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.*;
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.*;
import java.util.Objects;
import java.util.function.Supplier;
import javax.annotation.Nonnull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

final class NexusPollTask implements MultiThreadedPoller.PollTask<NexusTask> {
  private static final Logger log = LoggerFactory.getLogger(NexusPollTask.class);

  private final WorkflowServiceStubs service;
  private final TrackingSlotSupplier<NexusSlotInfo> slotSupplier;
  private final Scope metricsScope;
  private final PollNexusTaskQueueRequest pollRequest;
  private final PollerTracker pollerTracker;

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

  @SuppressWarnings("deprecation")
  public NexusPollTask(
      @Nonnull WorkflowServiceStubs service,
      @Nonnull String namespace,
      @Nonnull String taskQueue,
      @Nonnull String identity,
      @Nonnull String workerInstanceKey,
      @Nonnull WorkerVersioningOptions versioningOptions,
      @Nonnull TrackingSlotSupplier<NexusSlotInfo> slotSupplier,
      @Nonnull Scope metricsScope,
      @Nonnull Supplier<GetSystemInfoResponse.Capabilities> serverCapabilities,
      @Nonnull PollerTracker pollerTracker,
      boolean workerCommandsTaskQueue) {
    this.service = Objects.requireNonNull(service);
    this.slotSupplier = slotSupplier;
    this.metricsScope = Objects.requireNonNull(metricsScope);
    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 NexusTask poll() {
    if (log.isTraceEnabled()) {
      log.trace("poll request begin: " + pollRequest);
    }
    PollNexusTaskQueueResponse response;
    SlotPermit permit;
    SlotSupplierFuture future;
    boolean isSuccessful = false;
    try {
      future =
          slotSupplier.reserveSlot(
              new SlotReservationData(
                  pollRequest.getTaskQueue().getName(),
                  pollRequest.getIdentity(),
                  pollRequest.getWorkerVersionCapabilities().getBuildId()));
    } catch (Exception e) {
      log.warn("Error while trying to reserve a slot for a nexus task", e.getCause());
      return null;
    }
    permit = MultiThreadedPoller.getSlotPermitAndHandleInterrupts(future, slotSupplier);
    if (permit == null) return null;

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

    try {
      response =
          service
              .blockingStub()
              .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope)
              .pollNexusTaskQueue(pollRequest);

      if (response == null || response.getTaskToken().isEmpty()) {
        metricsScope.counter(MetricsType.NEXUS_POLL_NO_TASK_COUNTER).inc(1);
        return null;
      }

      Timestamp startedTime = ProtobufTimeUtils.getCurrentProtoTime();
      metricsScope
          .timer(MetricsType.NEXUS_SCHEDULE_TO_START_LATENCY)
          .record(
              ProtobufTimeUtils.toM3Duration(
                  startedTime, response.getRequest().getScheduledTime()));

      isSuccessful = true;
      pollerTracker.pollSucceeded();
      return new NexusTask(
          response,
          permit,
          () -> slotSupplier.releaseSlot(SlotReleaseReason.taskComplete(), permit));
    } finally {
      MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.NEXUS_TASK)
          .gauge(MetricsType.NUM_POLLERS)
          .update(pollerTracker.pollCompleted());

      if (!isSuccessful) slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit);
    }
  }
}

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.temporal.api.common.v1.WorkerVersionCapabilities;
8	import io.temporal.api.enums.v1.TaskQueueKind;
9	import io.temporal.api.taskqueue.v1.TaskQueue;
10	import io.temporal.api.workflowservice.v1.*;
11	import io.temporal.internal.common.ProtobufTimeUtils;
12	import io.temporal.serviceclient.MetricsTag;
13	import io.temporal.serviceclient.WorkflowServiceStubs;
14	import io.temporal.worker.MetricsType;
15	import io.temporal.worker.PollerTypeMetricsTag;
16	import io.temporal.worker.tuning.*;
17	import java.util.Objects;
18	import java.util.function.Supplier;
19	import javax.annotation.Nonnull;
20	import org.slf4j.Logger;
21	import org.slf4j.LoggerFactory;
22
23	final class NexusPollTask implements MultiThreadedPoller.PollTask<NexusTask> {
24	private static final Logger log = LoggerFactory.getLogger(NexusPollTask.class);	1✔
25
26	private final WorkflowServiceStubs service;
27	private final TrackingSlotSupplier<NexusSlotInfo> slotSupplier;
28	private final Scope metricsScope;
29	private final PollNexusTaskQueueRequest pollRequest;
30	private final PollerTracker pollerTracker;
31
32	@SuppressWarnings("deprecation")
33	public NexusPollTask(
34	@Nonnull WorkflowServiceStubs service,
35	@Nonnull String namespace,
36	@Nonnull String taskQueue,
37	@Nonnull String identity,
38	@Nonnull String workerInstanceKey,
39	@Nonnull WorkerVersioningOptions versioningOptions,
40	@Nonnull TrackingSlotSupplier<NexusSlotInfo> slotSupplier,
41	@Nonnull Scope metricsScope,
42	@Nonnull Supplier<GetSystemInfoResponse.Capabilities> serverCapabilities,
43	@Nonnull PollerTracker pollerTracker) {
44	this(	×
45	service,
46	namespace,
47	taskQueue,
48	identity,
49	workerInstanceKey,
50	versioningOptions,
51	slotSupplier,
52	metricsScope,
53	serverCapabilities,
54	pollerTracker,
55	false);
56	}	×
57
58	@SuppressWarnings("deprecation")
59	public NexusPollTask(
60	@Nonnull WorkflowServiceStubs service,
61	@Nonnull String namespace,
62	@Nonnull String taskQueue,
63	@Nonnull String identity,
64	@Nonnull String workerInstanceKey,
65	@Nonnull WorkerVersioningOptions versioningOptions,
66	@Nonnull TrackingSlotSupplier<NexusSlotInfo> slotSupplier,
67	@Nonnull Scope metricsScope,
68	@Nonnull Supplier<GetSystemInfoResponse.Capabilities> serverCapabilities,
69	@Nonnull PollerTracker pollerTracker,
70	boolean workerCommandsTaskQueue) {	1✔
71	this.service = Objects.requireNonNull(service);	1✔
72	this.slotSupplier = slotSupplier;	1✔
73	this.metricsScope = Objects.requireNonNull(metricsScope);	1✔
74	this.pollerTracker = Objects.requireNonNull(pollerTracker);	1✔
75
76	PollNexusTaskQueueRequest.Builder pollRequest =
77	PollNexusTaskQueueRequest.newBuilder()	1✔
78	.setNamespace(namespace)	1✔
79	.setIdentity(identity)	1✔
80	.setTaskQueue(	1✔
81	TaskQueue.newBuilder()	1✔
82	.setName(taskQueue)	1✔
83	.setKind(	1✔
84	workerCommandsTaskQueue	1!
85	? TaskQueueKind.TASK_QUEUE_KIND_WORKER_COMMANDS	×
86	: TaskQueueKind.TASK_QUEUE_KIND_NORMAL));	1✔
87	pollRequest.setWorkerInstanceKey(workerInstanceKey);	1✔
88
89	if (!workerCommandsTaskQueue) {	1!
90	if (versioningOptions.getWorkerDeploymentOptions() != null) {	1!
NEW 91	pollRequest.setDeploymentOptions(	×
NEW 92	WorkerVersioningProtoUtils.deploymentOptionsToProto(	×
NEW 93	versioningOptions.getWorkerDeploymentOptions()));	×
94	} else if (serverCapabilities.get().getBuildIdBasedVersioning()) {	1!
NEW 95	pollRequest.setWorkerVersionCapabilities(	×
NEW 96	WorkerVersionCapabilities.newBuilder()	×
NEW 97	.setBuildId(versioningOptions.getBuildId())	×
NEW 98	.setUseVersioning(versioningOptions.isUsingVersioning())	×
NEW 99	.build());	×
100	}
101	}
102	this.pollRequest = pollRequest.build();	1✔
103	}	1✔
104
105	@Override
106	@SuppressWarnings("deprecation")
107	public NexusTask poll() {
108	if (log.isTraceEnabled()) {	1!
109	log.trace("poll request begin: " + pollRequest);	×
110	}
111	PollNexusTaskQueueResponse response;
112	SlotPermit permit;
113	SlotSupplierFuture future;
114	boolean isSuccessful = false;	1✔
115	try {
116	future =	1✔
117	slotSupplier.reserveSlot(	1✔
118	new SlotReservationData(
119	pollRequest.getTaskQueue().getName(),	1✔
120	pollRequest.getIdentity(),	1✔
121	pollRequest.getWorkerVersionCapabilities().getBuildId()));	1✔
122	} catch (Exception e) {	×
123	log.warn("Error while trying to reserve a slot for a nexus task", e.getCause());	×
124	return null;	×
125	}	1✔
126	permit = MultiThreadedPoller.getSlotPermitAndHandleInterrupts(future, slotSupplier);	1✔
127	if (permit == null) return null;	1!
128
129	MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.NEXUS_TASK)	1✔
130	.gauge(MetricsType.NUM_POLLERS)	1✔
131	.update(pollerTracker.pollStarted());	1✔
132
133	try {
134	response =	1✔
135	service
136	.blockingStub()	1✔
137	.withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope)	1✔
138	.pollNexusTaskQueue(pollRequest);	1✔
139
140	if (response == null \|\| response.getTaskToken().isEmpty()) {	1!
141	metricsScope.counter(MetricsType.NEXUS_POLL_NO_TASK_COUNTER).inc(1);	1✔
142	return null;	1✔
143	}
144
145	Timestamp startedTime = ProtobufTimeUtils.getCurrentProtoTime();	1✔
146	metricsScope	1✔
147	.timer(MetricsType.NEXUS_SCHEDULE_TO_START_LATENCY)	1✔
148	.record(	1✔
149	ProtobufTimeUtils.toM3Duration(	1✔
150	startedTime, response.getRequest().getScheduledTime()));	1✔
151
152	isSuccessful = true;	1✔
153	pollerTracker.pollSucceeded();	1✔
154	return new NexusTask(	1✔
155	response,
156	permit,
157	() -> slotSupplier.releaseSlot(SlotReleaseReason.taskComplete(), permit));	1✔
158	} finally {
159	MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.NEXUS_TASK)	1✔
160	.gauge(MetricsType.NUM_POLLERS)	1✔
161	.update(pollerTracker.pollCompleted());	1✔
162
163	if (!isSuccessful) slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit);	1✔
164	}
165	}
166	}

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