#409

Committed 15 Oct 2026 11:35PM UTC coverage: 69.806% (+0.08%) from 69.728%

Build # #409

Build Type

push

github

Committed by

web-flow

Commit Message

Log workflow task duration with external storage metrics (#3140)

Coverage Stats

8001 of 13556 branches covered (59.02%)

Branch coverage included in aggregate %.

129 of 129 new or added lines in 7 files covered. (100.0%)

4 existing lines in 2 files now uncovered.

32368 of 44274 relevant lines covered (73.11%)

0.73 hits per line

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

55.88

/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClientImpl.java

package io.temporal.client;

import com.uber.m3.tally.Scope;
import io.temporal.api.common.v1.WorkflowExecution;
import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
import io.temporal.payload.context.ActivitySerializationContext;
import io.temporal.workflow.Functions;
import java.util.Optional;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

class ActivityCompletionClientImpl implements ActivityCompletionClient {

  private final ManualActivityCompletionClientFactory factory;
  private final Functions.Proc completionHandle;
  private final Scope metricsScope;
  private final @Nullable ActivitySerializationContext serializationContext;

  ActivityCompletionClientImpl(
      ManualActivityCompletionClientFactory manualActivityCompletionClientFactory,
      Functions.Proc completionHandle,
      Scope metricsScope,
      @Nullable ActivitySerializationContext serializationContext) {
    this.factory = manualActivityCompletionClientFactory;
    this.completionHandle = completionHandle;
    this.metricsScope = metricsScope;
    this.serializationContext = serializationContext;
  }

  @Override
  public <R> void complete(byte[] taskToken, R result) {
    try {
      factory.getClient(taskToken, metricsScope, serializationContext).complete(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <R> void complete(String workflowId, Optional<String> runId, String activityId, R result) {
    try {
      factory
          .getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)
          .complete(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <R> void completeStandalone(String activityId, Optional<String> activityRunId, R result) {
    try {
      factory
          .getClient(
              toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)
          .complete(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public void completeExceptionally(byte[] taskToken, Exception result) {
    try {
      factory.getClient(taskToken, metricsScope, serializationContext).fail(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public void completeExceptionally(
      String workflowId, Optional<String> runId, String activityId, Exception result) {
    try {
      factory
          .getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)
          .fail(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public void completeExceptionallyStandalone(
      String activityId, Optional<String> activityRunId, Exception result) {
    try {
      factory
          .getClient(
              toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)
          .fail(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <V> void reportCancellation(byte[] taskToken, V details) {
    try {
      factory.getClient(taskToken, metricsScope, serializationContext).reportCancellation(details);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <V> void reportCancellation(
      String workflowId, Optional<String> runId, String activityId, V details) {
    try {
      factory
          .getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)
          .reportCancellation(details);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <V> void reportCancellationStandalone(
      String activityId, Optional<String> activityRunId, V details) {
    try {
      factory
          .getClient(
              toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)
          .reportCancellation(details);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <V> void heartbeat(byte[] taskToken, V details) throws ActivityCompletionException {
    factory.getClient(taskToken, metricsScope).recordHeartbeat(details);
  }

  @Override
  public <V> void heartbeat(String workflowId, Optional<String> runId, String activityId, V details)
      throws ActivityCompletionException {
    factory
        .getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)
        .recordHeartbeat(details);
  }

  @Nonnull
  @Override
  public ActivityCompletionClient withContext(@Nonnull ActivitySerializationContext context) {
    return new ActivityCompletionClientImpl(factory, completionHandle, metricsScope, context);
  }

  @Override
  public <V> void heartbeatStandalone(String activityId, Optional<String> activityRunId, V details)
      throws ActivityCompletionException {
    factory
        .getClient(
            toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)
        .recordHeartbeat(details);
  }

  private static WorkflowExecution toExecution(String workflowId, Optional<String> runId) {
    return WorkflowExecution.newBuilder()
        .setWorkflowId(workflowId)
        .setRunId(runId.orElse(""))
        .build();
  }

  private static WorkflowExecution toStandaloneExecution(Optional<String> activityRunId) {
    // TODO: consider not using a WorkflowExecution here, but instead passing
    // workflow/activity and run IDs to the factory separately.
    return WorkflowExecution.newBuilder()
        .setWorkflowId("") // empty workflowId = standalone activity
        .setRunId(activityRunId.orElse(""))
        .build();
  }
}

1	package io.temporal.client;
2
3	import com.uber.m3.tally.Scope;
4	import io.temporal.api.common.v1.WorkflowExecution;
5	import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
6	import io.temporal.payload.context.ActivitySerializationContext;
7	import io.temporal.workflow.Functions;
8	import java.util.Optional;
9	import javax.annotation.Nonnull;
10	import javax.annotation.Nullable;
11
12	class ActivityCompletionClientImpl implements ActivityCompletionClient {
13
14	private final ManualActivityCompletionClientFactory factory;
15	private final Functions.Proc completionHandle;
16	private final Scope metricsScope;
17	private final @Nullable ActivitySerializationContext serializationContext;
18
19	ActivityCompletionClientImpl(
20	ManualActivityCompletionClientFactory manualActivityCompletionClientFactory,
21	Functions.Proc completionHandle,
22	Scope metricsScope,
23	@Nullable ActivitySerializationContext serializationContext) {	1 ✔
24	this.factory = manualActivityCompletionClientFactory;	1 ✔
25	this.completionHandle = completionHandle;	1 ✔
26	this.metricsScope = metricsScope;	1 ✔
27	this.serializationContext = serializationContext;	1 ✔
28	}	1 ✔
29
30	@Override
31	public <R> void complete(byte[] taskToken, R result) {
32	try {
33	factory.getClient(taskToken, metricsScope, serializationContext).complete(result);	1 ✔
34	} finally {
35	completionHandle.apply();	1 ✔
36	}
37	}	1 ✔
38
39	@Override
40	public <R> void complete(String workflowId, Optional<String> runId, String activityId, R result) {
41	try {
42	factory	×
43	.getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)	×
44	.complete(result);	×
45	} finally {
46	completionHandle.apply();	×
47	}
48	}	×
49
50	@Override
51	public <R> void completeStandalone(String activityId, Optional<String> activityRunId, R result) {
52	try {
53	factory	1 ✔
54	.getClient(	1 ✔
55	toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)	1 ✔
56	.complete(result);	1 ✔
57	} finally {
58	completionHandle.apply();	1 ✔
59	}
60	}	1 ✔
61
62	@Override
63	public void completeExceptionally(byte[] taskToken, Exception result) {
64	try {
65	factory.getClient(taskToken, metricsScope, serializationContext).fail(result);	×
66	} finally {
67	completionHandle.apply();	×
68	}
69	}	×
70
71	@Override
72	public void completeExceptionally(
73	String workflowId, Optional<String> runId, String activityId, Exception result) {
74	try {
75	factory	×
76	.getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)	×
77	.fail(result);	×
78	} finally {
79	completionHandle.apply();	×
80	}
81	}	×
82
83	@Override
84	public void completeExceptionallyStandalone(
85	String activityId, Optional<String> activityRunId, Exception result) {
86	try {
87	factory	1 ✔
88	.getClient(	1 ✔
89	toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)	1 ✔
90	.fail(result);	1 ✔
91	} finally {
92	completionHandle.apply();	1 ✔
93	}
94	}	1 ✔
95
96	@Override
97	public <V> void reportCancellation(byte[] taskToken, V details) {
98	try {
UNCOV 99	factory.getClient(taskToken, metricsScope, serializationContext).reportCancellation(details);	×
100	} finally {
UNCOV 101	completionHandle.apply();	×
102	}
UNCOV 103	}	×
104
105	@Override
106	public <V> void reportCancellation(
107	String workflowId, Optional<String> runId, String activityId, V details) {
108	try {
109	factory	×
110	.getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)	×
111	.reportCancellation(details);	×
112	} finally {
113	completionHandle.apply();	×
114	}
115	}	×
116
117	@Override
118	public <V> void reportCancellationStandalone(
119	String activityId, Optional<String> activityRunId, V details) {
120	try {
121	factory	1 ✔
122	.getClient(	1 ✔
123	toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)	1 ✔
124	.reportCancellation(details);	1 ✔
125	} finally {
126	completionHandle.apply();	1 ✔
127	}
128	}	1 ✔
129
130	@Override
131	public <V> void heartbeat(byte[] taskToken, V details) throws ActivityCompletionException {
132	factory.getClient(taskToken, metricsScope).recordHeartbeat(details);	1 ✔
133	}	1 ✔
134
135	@Override
136	public <V> void heartbeat(String workflowId, Optional<String> runId, String activityId, V details)
137	throws ActivityCompletionException {
138	factory	×
139	.getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)	×
140	.recordHeartbeat(details);	×
141	}	×
142
143	@Nonnull
144	@Override
145	public ActivityCompletionClient withContext(@Nonnull ActivitySerializationContext context) {
146	return new ActivityCompletionClientImpl(factory, completionHandle, metricsScope, context);	×
147	}
148
149	@Override
150	public <V> void heartbeatStandalone(String activityId, Optional<String> activityRunId, V details)
151	throws ActivityCompletionException {
152	factory	1 ✔
153	.getClient(	1 ✔
154	toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)	1 ✔
155	.recordHeartbeat(details);	1 ✔
156	}	1 ✔
157
158	private static WorkflowExecution toExecution(String workflowId, Optional<String> runId) {
159	return WorkflowExecution.newBuilder()	×
160	.setWorkflowId(workflowId)	×
161	.setRunId(runId.orElse(""))	×
162	.build();	×
163	}
164
165	private static WorkflowExecution toStandaloneExecution(Optional<String> activityRunId) {
166	// TODO: consider not using a WorkflowExecution here, but instead passing
167	// workflow/activity and run IDs to the factory separately.
168	return WorkflowExecution.newBuilder()	1 ✔
169	.setWorkflowId("") // empty workflowId = standalone activity	1 ✔
170	.setRunId(activityRunId.orElse(""))	1 ✔
171	.build();	1 ✔
172	}
173	}

temporalio / sdk-java / #409

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