#347

Committed 31 Jul 2026 06:59PM UTC coverage: 68.148% (-0.03%) from 68.181%

Build # #347

Build Type

push

github

Committed by

web-flow

Commit Message

NEXUS-485: Support Workflow Update as a Nexus Operation (#2945)

* NEXUS-485: Support Workflow Update as a Nexus Operation

* address comments, change signatures to newer

* address comments 2: add all workflow exec overloads

* address comments: log failed

Coverage Stats

7166 of 12554 branches covered (57.08%)

Branch coverage included in aggregate %.

118 of 296 new or added lines in 11 files covered. (39.86%)

28 existing lines in 7 files now uncovered.

29722 of 41575 relevant lines covered (71.49%)

0.71 hits per line

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

98.0

/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java

package io.temporal.internal.activity;

import com.uber.m3.tally.Scope;
import io.temporal.activity.ActivityExecutionContext;
import io.temporal.activity.ActivityInfo;
import io.temporal.activity.ManualActivityCompletionClient;
import io.temporal.client.ActivityCanceledException;
import io.temporal.client.ActivityCompletionException;
import io.temporal.client.WorkflowClient;
import io.temporal.common.CancellationToken;
import io.temporal.common.converter.DataConverter;
import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
import io.temporal.payload.context.ActivitySerializationContext;
import io.temporal.workflow.Functions;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Base implementation of an {@link ActivityExecutionContext}.
 *
 * @author fateev, suskin
 * @see ActivityExecutionContext
 */
@ThreadSafe
class ActivityExecutionContextImpl implements InternalActivityExecutionContext {
  private final Lock lock = new ReentrantLock();
  private final WorkflowClient client;
  private final Object activity;
  private final ManualActivityCompletionClientFactory manualCompletionClientFactory;
  private final Functions.Proc completionHandle;
  private final HeartbeatContext heartbeatContext;
  private final Functions.Proc closeCallback;

  private final Scope metricsScope;
  private final ActivityInfo info;
  private boolean useLocalManualCompletion;
  private boolean doNotCompleteOnReturn;

  /** Create an ActivityExecutionContextImpl with the given attributes. */
  ActivityExecutionContextImpl(
      WorkflowClient client,
      String namespace,
      Object activity,
      ActivityInfo info,
      DataConverter dataConverter,
      ScheduledExecutorService heartbeatExecutor,
      ManualActivityCompletionClientFactory manualCompletionClientFactory,
      Functions.Proc completionHandle,
      Scope metricsScope,
      String identity,
      Duration maxHeartbeatThrottleInterval,
      Duration defaultHeartbeatThrottleInterval,
      Functions.Proc closeCallback) {
    this.client = client;
    this.activity = activity;
    this.metricsScope = metricsScope;
    this.info = info;
    this.completionHandle = completionHandle;
    this.closeCallback = closeCallback;
    this.manualCompletionClientFactory = manualCompletionClientFactory;
    this.heartbeatContext =
        new HeartbeatContextImpl(
            client.getWorkflowServiceStubs(),
            namespace,
            info,
            dataConverter,
            heartbeatExecutor,
            metricsScope,
            identity,
            maxHeartbeatThrottleInterval,
            defaultHeartbeatThrottleInterval);
  }

  /**
   * @see ActivityExecutionContext#heartbeat(Object)
   */
  @Override
  public <V> void heartbeat(V details) throws ActivityCompletionException {
    heartbeatContext.heartbeat(details);
  }

  @Override
  public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass) {
    return getHeartbeatDetails(detailsClass, detailsClass);
  }

  @Override
  public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsGenericType) {
    return heartbeatContext.getHeartbeatDetails(detailsClass, detailsGenericType);
  }

  @Override
  public <V> Optional<V> getLastHeartbeatDetails(Class<V> detailsClass) {
    return getLastHeartbeatDetails(detailsClass, detailsClass);
  }

  @Override
  @SuppressWarnings("unchecked")
  public <V> Optional<V> getLastHeartbeatDetails(Class<V> detailsClass, Type detailsGenericType) {
    return heartbeatContext.getLastHeartbeatDetails(detailsClass, detailsGenericType);
  }

  @Override
  public byte[] getTaskToken() {
    return info.getTaskToken();
  }

  @Override
  public CancellationToken<ActivityCanceledException> getCancellationToken() {
    return heartbeatContext.getCancellationToken();
  }

  @Override
  public void doNotCompleteOnReturn() {
    lock.lock();
    try {
      doNotCompleteOnReturn = true;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public boolean isDoNotCompleteOnReturn() {
    lock.lock();
    try {
      return doNotCompleteOnReturn;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public boolean isUseLocalManualCompletion() {
    lock.lock();
    try {
      return useLocalManualCompletion;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public ManualActivityCompletionClient useLocalManualCompletion() {
    lock.lock();
    try {
      doNotCompleteOnReturn();
      useLocalManualCompletion = true;
      ActivitySerializationContext activitySerializationContext =
          new ActivitySerializationContext(info);
      return new CompletionAwareManualCompletionClient(
          manualCompletionClientFactory.getClient(
              info.getTaskToken(), metricsScope, activitySerializationContext),
          completionHandle);
    } finally {
      lock.unlock();
    }
  }

  @Override
  public Scope getMetricsScope() {
    return metricsScope;
  }

  @Override
  public ActivityInfo getInfo() {
    return info;
  }

  @Override
  public Object getLastHeartbeatValue() {
    return heartbeatContext.getLatestHeartbeatDetails();
  }

  @Override
  public void cancelOutstandingHeartbeat() {
    heartbeatContext.cancelOutstandingHeartbeat();
    closeCallback.apply();
  }

  @Override
  public void asyncCompletionStarted() {
    heartbeatContext.asyncCompletionStarted();
  }

  void cancelFromWorkerCommand() {
    heartbeatContext.cancelFromWorkerCommand();
  }

  @Override
  public WorkflowClient getWorkflowClient() {
    return client;
  }

  @Override
  public Object getInstance() {
    return activity;
  }
}

1	package io.temporal.internal.activity;
2
3	import com.uber.m3.tally.Scope;
4	import io.temporal.activity.ActivityExecutionContext;
5	import io.temporal.activity.ActivityInfo;
6	import io.temporal.activity.ManualActivityCompletionClient;
7	import io.temporal.client.ActivityCanceledException;
8	import io.temporal.client.ActivityCompletionException;
9	import io.temporal.client.WorkflowClient;
10	import io.temporal.common.CancellationToken;
11	import io.temporal.common.converter.DataConverter;
12	import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
13	import io.temporal.payload.context.ActivitySerializationContext;
14	import io.temporal.workflow.Functions;
15	import java.lang.reflect.Type;
16	import java.time.Duration;
17	import java.util.Optional;
18	import java.util.concurrent.ScheduledExecutorService;
19	import java.util.concurrent.locks.Lock;
20	import java.util.concurrent.locks.ReentrantLock;
21	import javax.annotation.concurrent.ThreadSafe;
22
23	/**
24	* Base implementation of an {@link ActivityExecutionContext}.
25	*
26	* @author fateev, suskin
27	* @see ActivityExecutionContext
28	*/
29	@ThreadSafe
30	class ActivityExecutionContextImpl implements InternalActivityExecutionContext {
31	private final Lock lock = new ReentrantLock();	1✔
32	private final WorkflowClient client;
33	private final Object activity;
34	private final ManualActivityCompletionClientFactory manualCompletionClientFactory;
35	private final Functions.Proc completionHandle;
36	private final HeartbeatContext heartbeatContext;
37	private final Functions.Proc closeCallback;
38
39	private final Scope metricsScope;
40	private final ActivityInfo info;
41	private boolean useLocalManualCompletion;
42	private boolean doNotCompleteOnReturn;
43
44	/** Create an ActivityExecutionContextImpl with the given attributes. */
45	ActivityExecutionContextImpl(
46	WorkflowClient client,
47	String namespace,
48	Object activity,
49	ActivityInfo info,
50	DataConverter dataConverter,
51	ScheduledExecutorService heartbeatExecutor,
52	ManualActivityCompletionClientFactory manualCompletionClientFactory,
53	Functions.Proc completionHandle,
54	Scope metricsScope,
55	String identity,
56	Duration maxHeartbeatThrottleInterval,
57	Duration defaultHeartbeatThrottleInterval,
58	Functions.Proc closeCallback) {	1✔
59	this.client = client;	1✔
60	this.activity = activity;	1✔
61	this.metricsScope = metricsScope;	1✔
62	this.info = info;	1✔
63	this.completionHandle = completionHandle;	1✔
64	this.closeCallback = closeCallback;	1✔
65	this.manualCompletionClientFactory = manualCompletionClientFactory;	1✔
66	this.heartbeatContext =	1✔
67	new HeartbeatContextImpl(
68	client.getWorkflowServiceStubs(),	1✔
69	namespace,
70	info,
71	dataConverter,
72	heartbeatExecutor,
73	metricsScope,
74	identity,
75	maxHeartbeatThrottleInterval,
76	defaultHeartbeatThrottleInterval);
77	}	1✔
78
79	/**
80	* @see ActivityExecutionContext#heartbeat(Object)
81	*/
82	@Override
83	public <V> void heartbeat(V details) throws ActivityCompletionException {
84	heartbeatContext.heartbeat(details);	1✔
85	}	1✔
86
87	@Override
88	public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass) {
89	return getHeartbeatDetails(detailsClass, detailsClass);	1✔
90	}
91
92	@Override
93	public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsGenericType) {
94	return heartbeatContext.getHeartbeatDetails(detailsClass, detailsGenericType);	1✔
95	}
96
97	@Override
98	public <V> Optional<V> getLastHeartbeatDetails(Class<V> detailsClass) {
99	return getLastHeartbeatDetails(detailsClass, detailsClass);	1✔
100	}
101
102	@Override
103	@SuppressWarnings("unchecked")
104	public <V> Optional<V> getLastHeartbeatDetails(Class<V> detailsClass, Type detailsGenericType) {
105	return heartbeatContext.getLastHeartbeatDetails(detailsClass, detailsGenericType);	1✔
106	}
107
108	@Override
109	public byte[] getTaskToken() {
UNCOV 110	return info.getTaskToken();	×
111	}
112
113	@Override
114	public CancellationToken<ActivityCanceledException> getCancellationToken() {
115	return heartbeatContext.getCancellationToken();	1✔
116	}
117
118	@Override
119	public void doNotCompleteOnReturn() {
120	lock.lock();	1✔
121	try {
122	doNotCompleteOnReturn = true;	1✔
123	} finally {
124	lock.unlock();	1✔
125	}
126	}	1✔
127
128	@Override
129	public boolean isDoNotCompleteOnReturn() {
130	lock.lock();	1✔
131	try {
132	return doNotCompleteOnReturn;	1✔
133	} finally {
134	lock.unlock();	1✔
135	}
136	}
137
138	@Override
139	public boolean isUseLocalManualCompletion() {
140	lock.lock();	1✔
141	try {
142	return useLocalManualCompletion;	1✔
143	} finally {
144	lock.unlock();	1✔
145	}
146	}
147
148	@Override
149	public ManualActivityCompletionClient useLocalManualCompletion() {
150	lock.lock();	1✔
151	try {
152	doNotCompleteOnReturn();	1✔
153	useLocalManualCompletion = true;	1✔
154	ActivitySerializationContext activitySerializationContext =	1✔
155	new ActivitySerializationContext(info);
156	return new CompletionAwareManualCompletionClient(	1✔
157	manualCompletionClientFactory.getClient(	1✔
158	info.getTaskToken(), metricsScope, activitySerializationContext),	1✔
159	completionHandle);
160	} finally {
161	lock.unlock();	1✔
162	}
163	}
164
165	@Override
166	public Scope getMetricsScope() {
167	return metricsScope;	1✔
168	}
169
170	@Override
171	public ActivityInfo getInfo() {
172	return info;	1✔
173	}
174
175	@Override
176	public Object getLastHeartbeatValue() {
177	return heartbeatContext.getLatestHeartbeatDetails();	1✔
178	}
179
180	@Override
181	public void cancelOutstandingHeartbeat() {
182	heartbeatContext.cancelOutstandingHeartbeat();	1✔
183	closeCallback.apply();	1✔
184	}	1✔
185
186	@Override
187	public void asyncCompletionStarted() {
188	heartbeatContext.asyncCompletionStarted();	1✔
189	}	1✔
190
191	void cancelFromWorkerCommand() {
192	heartbeatContext.cancelFromWorkerCommand();	1✔
193	}	1✔
194
195	@Override
196	public WorkflowClient getWorkflowClient() {
197	return client;	1✔
198	}
199
200	@Override
201	public Object getInstance() {
202	return activity;	1✔
203	}
204	}

temporalio / sdk-java / #347

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