#415

Committed 03 Nov 2026 09:28PM UTC coverage: 70.566% (+0.05%) from 70.518%

Build # #415

Build Type

push

github

Committed by

web-flow

Commit Message

Dedup retried Nexus signals (#3180)

Coverage Stats

8267 of 13774 branches covered (60.02%)

Branch coverage included in aggregate %.

46 of 46 new or added lines in 6 files covered. (100.0%)

5 existing lines in 4 files now uncovered.

32985 of 44685 relevant lines covered (73.82%)

0.74 hits per line

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

65.88

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

package io.temporal.internal.worker;

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

import com.google.protobuf.DoubleValue;
import com.uber.m3.tally.Scope;
import io.grpc.Context;
import io.temporal.api.common.v1.WorkerVersionCapabilities;
import io.temporal.api.taskqueue.v1.TaskQueue;
import io.temporal.api.taskqueue.v1.TaskQueueMetadata;
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.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.ActivitySlotInfo;
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 AsyncActivityPollTask implements AsyncPoller.PollTaskAsync<ActivityTask> {
  private static final Logger log = LoggerFactory.getLogger(AsyncActivityPollTask.class);

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

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

    PollActivityTaskQueueRequest.Builder pollRequest =
        PollActivityTaskQueueRequest.newBuilder()
            .setNamespace(namespace)
            .setIdentity(identity)
            .setTaskQueue(TaskQueue.newBuilder().setName(taskQueue));
    pollRequest.setWorkerInstanceKey(workerInstanceKey);
    if (workerControlTaskQueue != null) {
      pollRequest.setWorkerControlTaskQueue(workerControlTaskQueue);
    }
    if (activitiesPerSecond > 0) {
      pollRequest.setTaskQueueMetadata(
          TaskQueueMetadata.newBuilder()
              .setMaxTasksPerSecond(DoubleValue.newBuilder().setValue(activitiesPerSecond).build())
              .build());
    }

    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
  public CompletableFuture<ActivityTask> poll(SlotPermit permit) {
    if (log.isTraceEnabled()) {
      log.trace("poll request begin: " + pollRequest);
    }

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

    CompletableFuture<PollActivityTaskQueueResponse> response = null;
    try {
      response =
          grpcContext.call(
              () ->
                  GrpcUtils.toCompletableFuture(
                      service
                          .futureStub()
                          .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope)
                          .pollActivityTaskQueue(pollRequest)));
    } catch (Exception e) {
      MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.ACTIVITY_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.ACTIVITY_POLL_NO_TASK_COUNTER).inc(1);
                return null;
              }
              pollerTracker.pollSucceeded();
              metricsScope
                  .timer(MetricsType.ACTIVITY_SCHEDULE_TO_START_LATENCY)
                  .record(
                      ProtobufTimeUtils.toM3Duration(
                          r.getStartedTime(), r.getCurrentAttemptScheduledTime()));
              return new ActivityTask(
                  r,
                  permit,
                  () -> slotSupplier.releaseSlot(SlotReleaseReason.taskComplete(), permit));
            })
        .whenComplete(
            (r, e) -> {
              MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.ACTIVITY_TASK)
                  .gauge(MetricsType.NUM_POLLERS)
                  .update(pollerTracker.pollCompleted());
            });
  }

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

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

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

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.DoubleValue;
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.taskqueue.v1.TaskQueue;
10	import io.temporal.api.taskqueue.v1.TaskQueueMetadata;
11	import io.temporal.api.workflowservice.v1.GetSystemInfoResponse;
12	import io.temporal.api.workflowservice.v1.PollActivityTaskQueueRequest;
13	import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse;
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.ActivitySlotInfo;
21	import io.temporal.worker.tuning.SlotPermit;
22	import io.temporal.worker.tuning.SlotReleaseReason;
23	import java.util.Objects;
24	import java.util.concurrent.CompletableFuture;
25	import java.util.function.Supplier;
26	import javax.annotation.Nonnull;
27	import org.slf4j.Logger;
28	import org.slf4j.LoggerFactory;
29
30	public class AsyncActivityPollTask implements AsyncPoller.PollTaskAsync<ActivityTask> {
31	private static final Logger log = LoggerFactory.getLogger(AsyncActivityPollTask.class);	1 ✔
32
33	private final TrackingSlotSupplier<?> slotSupplier;
34	private final WorkflowServiceStubs service;
35	private final Scope metricsScope;
36	private final PollActivityTaskQueueRequest pollRequest;
37	private final Context.CancellableContext grpcContext = Context.ROOT.withCancellation();	1 ✔
38	private final PollerTracker pollerTracker;
39
40	@SuppressWarnings("deprecation")
41	public AsyncActivityPollTask(
42	@Nonnull WorkflowServiceStubs service,
43	@Nonnull String namespace,
44	@Nonnull String taskQueue,
45	@Nonnull String identity,
46	@Nonnull String workerInstanceKey,
47	@Nonnull WorkerVersioningOptions versioningOptions,
48	double activitiesPerSecond,
49	@Nonnull TrackingSlotSupplier<ActivitySlotInfo> slotSupplier,
50	@Nonnull Scope metricsScope,
51	@Nonnull Supplier<GetSystemInfoResponse.Capabilities> serverCapabilities,
52	@Nonnull PollerTracker pollerTracker,
53	String workerControlTaskQueue) {	1 ✔
54	this.service = service;	1 ✔
55	this.slotSupplier = slotSupplier;	1 ✔
56	this.metricsScope = metricsScope;	1 ✔
57	this.pollerTracker = Objects.requireNonNull(pollerTracker);	1 ✔
58
59	PollActivityTaskQueueRequest.Builder pollRequest =
60	PollActivityTaskQueueRequest.newBuilder()	1 ✔
61	.setNamespace(namespace)	1 ✔
62	.setIdentity(identity)	1 ✔
63	.setTaskQueue(TaskQueue.newBuilder().setName(taskQueue));	1 ✔
64	pollRequest.setWorkerInstanceKey(workerInstanceKey);	1 ✔
65	if (workerControlTaskQueue != null) {	1 !
66	pollRequest.setWorkerControlTaskQueue(workerControlTaskQueue);	×
67	}
68	if (activitiesPerSecond > 0) {	1 !
69	pollRequest.setTaskQueueMetadata(	×
70	TaskQueueMetadata.newBuilder()	×
71	.setMaxTasksPerSecond(DoubleValue.newBuilder().setValue(activitiesPerSecond).build())	×
72	.build());	×
73	}
74
75	if (versioningOptions.getWorkerDeploymentOptions() != null) {	1 !
76	pollRequest.setDeploymentOptions(	×
77	WorkerVersioningProtoUtils.deploymentOptionsToProto(	×
78	versioningOptions.getWorkerDeploymentOptions()));	×
79	} else if (serverCapabilities.get().getBuildIdBasedVersioning()) {	1 !
80	pollRequest.setWorkerVersionCapabilities(	×
81	WorkerVersionCapabilities.newBuilder()	×
82	.setBuildId(versioningOptions.getBuildId())	×
83	.setUseVersioning(versioningOptions.isUsingVersioning())	×
84	.build());	×
85	}
86	this.pollRequest = pollRequest.build();	1 ✔
87	}	1 ✔
88
89	@Override
90	public CompletableFuture<ActivityTask> poll(SlotPermit permit) {
91	if (log.isTraceEnabled()) {	1 !
92	log.trace("poll request begin: " + pollRequest);	×
93	}
94
95	MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.ACTIVITY_TASK)	1 ✔
96	.gauge(MetricsType.NUM_POLLERS)	1 ✔
97	.update(pollerTracker.pollStarted());	1 ✔
98
99	CompletableFuture<PollActivityTaskQueueResponse> response = null;	1 ✔
100	try {
101	response =	1 ✔
102	grpcContext.call(	1 ✔
103	() ->
104	GrpcUtils.toCompletableFuture(	1 ✔
105	service
106	.futureStub()	1 ✔
107	.withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope)	1 ✔
108	.pollActivityTaskQueue(pollRequest)));	1 ✔
109	} catch (Exception e) {	×
110	MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.ACTIVITY_TASK)	×
111	.gauge(MetricsType.NUM_POLLERS)	×
112	.update(pollerTracker.pollCompleted());	×
113	throw new RuntimeException(e);	×
114	}	1 ✔
115
116	return response	1 ✔
117	.thenApply(	1 ✔
118	r -> {
119	if (r == null \|\| r.getTaskToken().isEmpty()) {	1 !
UNCOV 120	metricsScope.counter(MetricsType.ACTIVITY_POLL_NO_TASK_COUNTER).inc(1);	×
UNCOV 121	return null;	×
122	}
123	pollerTracker.pollSucceeded();	1 ✔
124	metricsScope	1 ✔
125	.timer(MetricsType.ACTIVITY_SCHEDULE_TO_START_LATENCY)	1 ✔
126	.record(	1 ✔
127	ProtobufTimeUtils.toM3Duration(	1 ✔
128	r.getStartedTime(), r.getCurrentAttemptScheduledTime()));	1 ✔
129	return new ActivityTask(	1 ✔
130	r,
131	permit,
132	() -> slotSupplier.releaseSlot(SlotReleaseReason.taskComplete(), permit));	1 ✔
133	})
134	.whenComplete(	1 ✔
135	(r, e) -> {
136	MetricsTag.tagged(metricsScope, PollerTypeMetricsTag.PollerType.ACTIVITY_TASK)	1 ✔
137	.gauge(MetricsType.NUM_POLLERS)	1 ✔
138	.update(pollerTracker.pollCompleted());	1 ✔
139	});	1 ✔
140	}
141
142	@Override
143	public void cancel(Throwable cause) {
144	grpcContext.cancel(cause);	1 ✔
145	}	1 ✔
146
147	@Override
148	public String getLabel() {
149	return "AsyncActivityPollTask";	1 ✔
150	}
151
152	@Override
153	public String toString() {
154	return "AsyncActivityPollTask{}";	×
155	}
156	}

temporalio / sdk-java / #415

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