#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

92.59

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

package io.temporal.internal.activity;

import com.uber.m3.tally.Scope;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.temporal.activity.ActivityExecutionContext;
import io.temporal.activity.ActivityInfo;
import io.temporal.api.common.v1.Payloads;
import io.temporal.api.enums.v1.TimeoutType;
import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse;
import io.temporal.client.*;
import io.temporal.common.CancellationToken;
import io.temporal.common.converter.DataConverter;
import io.temporal.failure.TimeoutFailure;
import io.temporal.internal.client.ActivityClientHelper;
import io.temporal.internal.concurrent.structured.CancelSource;
import io.temporal.payload.context.ActivitySerializationContext;
import io.temporal.serviceclient.WorkflowServiceStubs;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.annotation.concurrent.ThreadSafe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@ThreadSafe
class HeartbeatContextImpl implements HeartbeatContext {
  private static final Logger log = LoggerFactory.getLogger(HeartbeatContextImpl.class);
  private static final long HEARTBEAT_RETRY_WAIT_MILLIS = 1000;
  // Buffer added to the heartbeat timeout to avoid racing with the server's own timeout tracking.
  private static final long DEFAULT_LOCAL_HEARTBEAT_TIMEOUT_BUFFER_MILLIS = 5000;
  static final String LOCAL_TIMEOUT_BUFFER_PROPERTY = "temporal.activity.localTimeoutBufferMs";

  static long getLocalHeartbeatTimeoutBufferMillis() {
    String val = System.getProperty(LOCAL_TIMEOUT_BUFFER_PROPERTY);
    if (val != null) {
      try {
        return Long.parseLong(val);
      } catch (NumberFormatException e) {
        log.warn("Invalid {} value: {}", LOCAL_TIMEOUT_BUFFER_PROPERTY, val);
      }
    }
    return DEFAULT_LOCAL_HEARTBEAT_TIMEOUT_BUFFER_MILLIS;
  }

  private final Lock lock = new ReentrantLock();

  private final WorkflowServiceStubs service;
  private final String namespace;
  private final ActivityInfo info;
  private final String identity;
  private final ScheduledExecutorService heartbeatExecutor;
  private final long heartbeatIntervalMillis;
  private final DataConverter dataConverter;
  private final DataConverter dataConverterWithActivityContext;

  private final Scope metricsScope;
  private final Optional<Payloads> prevAttemptHeartbeatDetails;
  private final long heartbeatTimeoutMillis;
  private final long localHeartbeatTimeoutBufferMillis;

  // turned into true on a reception of the first heartbeat
  private boolean receivedAHeartbeat = false;
  private Object lastDetails;
  private boolean hasOutstandingHeartbeat;
  private ScheduledFuture<?> scheduledHeartbeat;

  // Deadline (in nanos, from System.nanoTime()) by which a successful heartbeat must occur.
  // 0 means no local timeout is active.
  private long heartbeatTimeoutDeadlineNanos;
  private boolean heartbeatTimedOut;
  private boolean rejectNewHeartbeats;

  private ActivityCompletionException lastException;
  private final CancelSource<ActivityCanceledException> cancellationSource =
      new CancelSource<>(ActivityCanceledException::new);

  public HeartbeatContextImpl(
      WorkflowServiceStubs service,
      String namespace,
      ActivityInfo info,
      DataConverter dataConverter,
      ScheduledExecutorService heartbeatExecutor,
      Scope metricsScope,
      String identity,
      Duration maxHeartbeatThrottleInterval,
      Duration defaultHeartbeatThrottleInterval) {
    this(
        service,
        namespace,
        info,
        dataConverter,
        heartbeatExecutor,
        metricsScope,
        identity,
        maxHeartbeatThrottleInterval,
        defaultHeartbeatThrottleInterval,
        getLocalHeartbeatTimeoutBufferMillis());
  }

  HeartbeatContextImpl(
      WorkflowServiceStubs service,
      String namespace,
      ActivityInfo info,
      DataConverter dataConverter,
      ScheduledExecutorService heartbeatExecutor,
      Scope metricsScope,
      String identity,
      Duration maxHeartbeatThrottleInterval,
      Duration defaultHeartbeatThrottleInterval,
      long localHeartbeatTimeoutBufferMillis) {
    this.service = service;
    this.metricsScope = metricsScope;
    this.dataConverter = dataConverter;
    this.dataConverterWithActivityContext =
        dataConverter.withContext(
            new ActivitySerializationContext(
                namespace,
                info.getWorkflowId(),
                info.getWorkflowType(),
                info.getActivityType(),
                info.getActivityTaskQueue(),
                info.isLocal()));
    this.namespace = namespace;
    this.info = info;
    this.identity = identity;
    this.prevAttemptHeartbeatDetails = info.getHeartbeatDetails();
    this.heartbeatExecutor = heartbeatExecutor;
    this.heartbeatIntervalMillis =
        getHeartbeatIntervalMs(
            info.getHeartbeatTimeout(),
            maxHeartbeatThrottleInterval,
            defaultHeartbeatThrottleInterval);
    this.heartbeatTimeoutMillis = info.getHeartbeatTimeout().toMillis();
    this.localHeartbeatTimeoutBufferMillis = localHeartbeatTimeoutBufferMillis;
    if (this.heartbeatTimeoutMillis > 0) {
      this.heartbeatTimeoutDeadlineNanos = computeHeartbeatTimeoutDeadlineNanos();
    }
  }

  /**
   * @see ActivityExecutionContext#heartbeat(Object)
   */
  @Override
  public <V> void heartbeat(V details) throws ActivityCompletionException {
    if (heartbeatExecutor.isShutdown()) {
      throw new ActivityWorkerShutdownException(info);
    }
    lock.lock();
    try {
      checkHeartbeatTimeoutDeadlineLocked();
      if (rejectNewHeartbeats) {
        throw new IllegalStateException("Cannot record heartbeats after async activity completion");
      }
      receivedAHeartbeat = true;
      lastDetails = details;
      hasOutstandingHeartbeat = true;
      // Only do sync heartbeat if there is no such call scheduled.
      if (scheduledHeartbeat == null) {
        doHeartBeatLocked(details);
      }
      if (lastException != null) {
        throw lastException;
      }
      cancellationSource.token().throwIfCancellationRequested();
    } finally {
      lock.unlock();
    }
  }

  /**
   * @see ActivityExecutionContext#getHeartbeatDetails(Class, Type)
   */
  @Override
  @SuppressWarnings("unchecked")
  public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsGenericType) {
    lock.lock();
    try {
      if (receivedAHeartbeat) {
        return Optional.ofNullable((V) this.lastDetails);
      } else {
        return Optional.ofNullable(
            dataConverterWithActivityContext.fromPayloads(
                0, prevAttemptHeartbeatDetails, detailsClass, detailsGenericType));
      }
    } finally {
      lock.unlock();
    }
  }

  /**
   * @see ActivityExecutionContext#getLastHeartbeatDetails(Class, Type)
   */
  @Override
  @SuppressWarnings("unchecked")
  public <V> Optional<V> getLastHeartbeatDetails(Class<V> detailsClass, Type detailsGenericType) {
    lock.lock();
    try {
      return Optional.ofNullable(
          dataConverterWithActivityContext.fromPayloads(
              0, prevAttemptHeartbeatDetails, detailsClass, detailsGenericType));
    } finally {
      lock.unlock();
    }
  }

  @Override
  public Object getLatestHeartbeatDetails() {
    lock.lock();
    try {
      if (receivedAHeartbeat) {
        return this.lastDetails;
      }
      return null;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public void cancelOutstandingHeartbeat() {
    lock.lock();
    try {
      if (scheduledHeartbeat != null) {
        scheduledHeartbeat.cancel(false);
        scheduledHeartbeat = null;
      }
      heartbeatTimeoutDeadlineNanos = 0;
      hasOutstandingHeartbeat = false;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public void cancelFromWorkerCommand() {
    lock.lock();
    try {
      requestCancelLocked();
    } finally {
      lock.unlock();
    }
  }

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

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

  private void doHeartBeatLocked(Object details) {
    long nextHeartbeatDelay;
    try {
      sendHeartbeatRequest(details);
      hasOutstandingHeartbeat = false;
      nextHeartbeatDelay = heartbeatIntervalMillis;
      // Reset the local heartbeat timeout deadline only on successful send.
      // If sends keep failing, the next heartbeat() call after the deadline will cancel the
      // activity.
      if (heartbeatTimeoutDeadlineNanos != 0) {
        heartbeatTimeoutDeadlineNanos = computeHeartbeatTimeoutDeadlineNanos();
      }
    } catch (StatusRuntimeException e) {
      // Not rethrowing to not fail activity implementation on intermittent connection or Temporal
      // errors.
      log.warn("Heartbeat failed", e);
      nextHeartbeatDelay = HEARTBEAT_RETRY_WAIT_MILLIS;
    } catch (Exception e) {
      log.error("Unexpected exception", e);
      nextHeartbeatDelay = HEARTBEAT_RETRY_WAIT_MILLIS;
    }

    scheduleNextHeartbeatLocked(nextHeartbeatDelay);
  }

  private void scheduleNextHeartbeatLocked(long delay) {
    scheduledHeartbeat =
        heartbeatExecutor.schedule(
            () -> {
              lock.lock();
              try {
                if (hasOutstandingHeartbeat) {
                  doHeartBeatLocked(lastDetails);
                } else {
                  // if no new heartbeats have been submitted in the previous time interval, we
                  // don't need to throttle
                  // and the next heartbeat should go immediately without following a schedule.
                  scheduledHeartbeat = null;
                }
              } finally {
                lock.unlock();
              }
            },
            delay,
            TimeUnit.MILLISECONDS);
  }

  private long computeHeartbeatTimeoutDeadlineNanos() {
    return System.nanoTime()
        + TimeUnit.MILLISECONDS.toNanos(heartbeatTimeoutMillis + localHeartbeatTimeoutBufferMillis);
  }

  private void checkHeartbeatTimeoutDeadlineLocked() {
    if (heartbeatTimedOut) {
      throw new ActivityCanceledException(
          info, new TimeoutFailure(null, null, TimeoutType.TIMEOUT_TYPE_HEARTBEAT));
    }
    if (heartbeatTimeoutDeadlineNanos != 0 && System.nanoTime() >= heartbeatTimeoutDeadlineNanos) {
      heartbeatTimedOut = true;
      log.warn(
          "Activity heartbeat timed out locally. ActivityId={}, activityType={}",
          info.getActivityId(),
          info.getActivityType());
      throw new ActivityCanceledException(
          info, new TimeoutFailure(null, null, TimeoutType.TIMEOUT_TYPE_HEARTBEAT));
    }
  }

  private void sendHeartbeatRequest(Object details) {
    try {
      RecordActivityTaskHeartbeatResponse status =
          ActivityClientHelper.sendHeartbeatRequest(
              service,
              namespace,
              identity,
              info.getTaskToken(),
              dataConverterWithActivityContext.toPayloads(details),
              metricsScope);
      if (status.getCancelRequested()) {
        requestCancelLocked();
      } else if (status.getActivityReset()) {
        lastException = new ActivityResetException(info);
      } else if (status.getActivityPaused()) {
        lastException = new ActivityPausedException(info);
      } else {
        lastException = null;
      }
    } catch (StatusRuntimeException e) {
      if (e.getStatus().getCode() == Status.Code.NOT_FOUND) {
        lastException = new ActivityNotExistsException(info, e);
      } else if (e.getStatus().getCode() == Status.Code.INVALID_ARGUMENT
          || e.getStatus().getCode() == Status.Code.FAILED_PRECONDITION) {
        lastException = new ActivityCompletionFailureException(info, e);
      } else {
        throw e;
      }
    }
  }

  private void requestCancelLocked() {
    ActivityCanceledException exception = new ActivityCanceledException(info);
    lastException = exception;
    cancellationSource.cancel(exception);
  }

  private static long getHeartbeatIntervalMs(
      Duration activityHeartbeatTimeout,
      Duration maxHeartbeatThrottleInterval,
      Duration defaultHeartbeatThrottleInterval) {
    long interval =
        activityHeartbeatTimeout.isZero()
            ? defaultHeartbeatThrottleInterval.toMillis()
            : (long) (0.8 * activityHeartbeatTimeout.toMillis());
    return Math.min(interval, maxHeartbeatThrottleInterval.toMillis());
  }
}

1	package io.temporal.internal.activity;
2
3	import com.uber.m3.tally.Scope;
4	import io.grpc.Status;
5	import io.grpc.StatusRuntimeException;
6	import io.temporal.activity.ActivityExecutionContext;
7	import io.temporal.activity.ActivityInfo;
8	import io.temporal.api.common.v1.Payloads;
9	import io.temporal.api.enums.v1.TimeoutType;
10	import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse;
11	import io.temporal.client.*;
12	import io.temporal.common.CancellationToken;
13	import io.temporal.common.converter.DataConverter;
14	import io.temporal.failure.TimeoutFailure;
15	import io.temporal.internal.client.ActivityClientHelper;
16	import io.temporal.internal.concurrent.structured.CancelSource;
17	import io.temporal.payload.context.ActivitySerializationContext;
18	import io.temporal.serviceclient.WorkflowServiceStubs;
19	import java.lang.reflect.Type;
20	import java.time.Duration;
21	import java.util.Optional;
22	import java.util.concurrent.ScheduledExecutorService;
23	import java.util.concurrent.ScheduledFuture;
24	import java.util.concurrent.TimeUnit;
25	import java.util.concurrent.locks.Lock;
26	import java.util.concurrent.locks.ReentrantLock;
27	import javax.annotation.concurrent.ThreadSafe;
28	import org.slf4j.Logger;
29	import org.slf4j.LoggerFactory;
30
31	@ThreadSafe
32	class HeartbeatContextImpl implements HeartbeatContext {
33	private static final Logger log = LoggerFactory.getLogger(HeartbeatContextImpl.class);	1✔
34	private static final long HEARTBEAT_RETRY_WAIT_MILLIS = 1000;
35	// Buffer added to the heartbeat timeout to avoid racing with the server's own timeout tracking.
36	private static final long DEFAULT_LOCAL_HEARTBEAT_TIMEOUT_BUFFER_MILLIS = 5000;
37	static final String LOCAL_TIMEOUT_BUFFER_PROPERTY = "temporal.activity.localTimeoutBufferMs";
38
39	static long getLocalHeartbeatTimeoutBufferMillis() {
40	String val = System.getProperty(LOCAL_TIMEOUT_BUFFER_PROPERTY);	1✔
41	if (val != null) {	1!
42	try {
43	return Long.parseLong(val);	×
44	} catch (NumberFormatException e) {	×
UNCOV 45	log.warn("Invalid {} value: {}", LOCAL_TIMEOUT_BUFFER_PROPERTY, val);	×
46	}
47	}
48	return DEFAULT_LOCAL_HEARTBEAT_TIMEOUT_BUFFER_MILLIS;	1✔
49	}
50
51	private final Lock lock = new ReentrantLock();	1✔
52
53	private final WorkflowServiceStubs service;
54	private final String namespace;
55	private final ActivityInfo info;
56	private final String identity;
57	private final ScheduledExecutorService heartbeatExecutor;
58	private final long heartbeatIntervalMillis;
59	private final DataConverter dataConverter;
60	private final DataConverter dataConverterWithActivityContext;
61
62	private final Scope metricsScope;
63	private final Optional<Payloads> prevAttemptHeartbeatDetails;
64	private final long heartbeatTimeoutMillis;
65	private final long localHeartbeatTimeoutBufferMillis;
66
67	// turned into true on a reception of the first heartbeat
68	private boolean receivedAHeartbeat = false;	1✔
69	private Object lastDetails;
70	private boolean hasOutstandingHeartbeat;
71	private ScheduledFuture<?> scheduledHeartbeat;
72
73	// Deadline (in nanos, from System.nanoTime()) by which a successful heartbeat must occur.
74	// 0 means no local timeout is active.
75	private long heartbeatTimeoutDeadlineNanos;
76	private boolean heartbeatTimedOut;
77	private boolean rejectNewHeartbeats;
78
79	private ActivityCompletionException lastException;
80	private final CancelSource<ActivityCanceledException> cancellationSource =	1✔
81	new CancelSource<>(ActivityCanceledException::new);
82
83	public HeartbeatContextImpl(
84	WorkflowServiceStubs service,
85	String namespace,
86	ActivityInfo info,
87	DataConverter dataConverter,
88	ScheduledExecutorService heartbeatExecutor,
89	Scope metricsScope,
90	String identity,
91	Duration maxHeartbeatThrottleInterval,
92	Duration defaultHeartbeatThrottleInterval) {
93	this(	1✔
94	service,
95	namespace,
96	info,
97	dataConverter,
98	heartbeatExecutor,
99	metricsScope,
100	identity,
101	maxHeartbeatThrottleInterval,
102	defaultHeartbeatThrottleInterval,
103	getLocalHeartbeatTimeoutBufferMillis());	1✔
104	}	1✔
105
106	HeartbeatContextImpl(
107	WorkflowServiceStubs service,
108	String namespace,
109	ActivityInfo info,
110	DataConverter dataConverter,
111	ScheduledExecutorService heartbeatExecutor,
112	Scope metricsScope,
113	String identity,
114	Duration maxHeartbeatThrottleInterval,
115	Duration defaultHeartbeatThrottleInterval,
116	long localHeartbeatTimeoutBufferMillis) {	1✔
117	this.service = service;	1✔
118	this.metricsScope = metricsScope;	1✔
119	this.dataConverter = dataConverter;	1✔
120	this.dataConverterWithActivityContext =	1✔
121	dataConverter.withContext(	1✔
122	new ActivitySerializationContext(
123	namespace,
124	info.getWorkflowId(),	1✔
125	info.getWorkflowType(),	1✔
126	info.getActivityType(),	1✔
127	info.getActivityTaskQueue(),	1✔
128	info.isLocal()));	1✔
129	this.namespace = namespace;	1✔
130	this.info = info;	1✔
131	this.identity = identity;	1✔
132	this.prevAttemptHeartbeatDetails = info.getHeartbeatDetails();	1✔
133	this.heartbeatExecutor = heartbeatExecutor;	1✔
134	this.heartbeatIntervalMillis =	1✔
135	getHeartbeatIntervalMs(	1✔
136	info.getHeartbeatTimeout(),	1✔
137	maxHeartbeatThrottleInterval,
138	defaultHeartbeatThrottleInterval);
139	this.heartbeatTimeoutMillis = info.getHeartbeatTimeout().toMillis();	1✔
140	this.localHeartbeatTimeoutBufferMillis = localHeartbeatTimeoutBufferMillis;	1✔
141	if (this.heartbeatTimeoutMillis > 0) {	1✔
142	this.heartbeatTimeoutDeadlineNanos = computeHeartbeatTimeoutDeadlineNanos();	1✔
143	}
144	}	1✔
145
146	/**
147	* @see ActivityExecutionContext#heartbeat(Object)
148	*/
149	@Override
150	public <V> void heartbeat(V details) throws ActivityCompletionException {
151	if (heartbeatExecutor.isShutdown()) {	1✔
152	throw new ActivityWorkerShutdownException(info);	1✔
153	}
154	lock.lock();	1✔
155	try {
156	checkHeartbeatTimeoutDeadlineLocked();	1✔
157	if (rejectNewHeartbeats) {	1✔
158	throw new IllegalStateException("Cannot record heartbeats after async activity completion");	1✔
159	}
160	receivedAHeartbeat = true;	1✔
161	lastDetails = details;	1✔
162	hasOutstandingHeartbeat = true;	1✔
163	// Only do sync heartbeat if there is no such call scheduled.
164	if (scheduledHeartbeat == null) {	1✔
165	doHeartBeatLocked(details);	1✔
166	}
167	if (lastException != null) {	1✔
168	throw lastException;	1✔
169	}
170	cancellationSource.token().throwIfCancellationRequested();	1✔
171	} finally {
172	lock.unlock();	1✔
173	}
174	}	1✔
175
176	/**
177	* @see ActivityExecutionContext#getHeartbeatDetails(Class, Type)
178	*/
179	@Override
180	@SuppressWarnings("unchecked")
181	public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsGenericType) {
182	lock.lock();	1✔
183	try {
184	if (receivedAHeartbeat) {	1✔
185	return Optional.ofNullable((V) this.lastDetails);	1✔
186	} else {
187	return Optional.ofNullable(	1✔
188	dataConverterWithActivityContext.fromPayloads(	1✔
189	0, prevAttemptHeartbeatDetails, detailsClass, detailsGenericType));
190	}
191	} finally {
192	lock.unlock();	1✔
193	}
194	}
195
196	/**
197	* @see ActivityExecutionContext#getLastHeartbeatDetails(Class, Type)
198	*/
199	@Override
200	@SuppressWarnings("unchecked")
201	public <V> Optional<V> getLastHeartbeatDetails(Class<V> detailsClass, Type detailsGenericType) {
202	lock.lock();	1✔
203	try {
204	return Optional.ofNullable(	1✔
205	dataConverterWithActivityContext.fromPayloads(	1✔
206	0, prevAttemptHeartbeatDetails, detailsClass, detailsGenericType));
207	} finally {
208	lock.unlock();	1✔
209	}
210	}
211
212	@Override
213	public Object getLatestHeartbeatDetails() {
214	lock.lock();	1✔
215	try {
216	if (receivedAHeartbeat) {	1✔
217	return this.lastDetails;	1✔
218	}
219	return null;	1✔
220	} finally {
221	lock.unlock();	1✔
222	}
223	}
224
225	@Override
226	public void cancelOutstandingHeartbeat() {
227	lock.lock();	1✔
228	try {
229	if (scheduledHeartbeat != null) {	1✔
230	scheduledHeartbeat.cancel(false);	1✔
231	scheduledHeartbeat = null;	1✔
232	}
233	heartbeatTimeoutDeadlineNanos = 0;	1✔
234	hasOutstandingHeartbeat = false;	1✔
235	} finally {
236	lock.unlock();	1✔
237	}
238	}	1✔
239
240	@Override
241	public void cancelFromWorkerCommand() {
242	lock.lock();	1✔
243	try {
244	requestCancelLocked();	1✔
245	} finally {
246	lock.unlock();	1✔
247	}
248	}	1✔
249
250	@Override
251	public void asyncCompletionStarted() {
252	lock.lock();	1✔
253	try {
254	rejectNewHeartbeats = true;	1✔
255	} finally {
256	lock.unlock();	1✔
257	}
258	}	1✔
259
260	@Override
261	public CancellationToken<ActivityCanceledException> getCancellationToken() {
262	return cancellationSource.token();	1✔
263	}
264
265	private void doHeartBeatLocked(Object details) {
266	long nextHeartbeatDelay;
267	try {
268	sendHeartbeatRequest(details);	1✔
269	hasOutstandingHeartbeat = false;	1✔
270	nextHeartbeatDelay = heartbeatIntervalMillis;	1✔
271	// Reset the local heartbeat timeout deadline only on successful send.
272	// If sends keep failing, the next heartbeat() call after the deadline will cancel the
273	// activity.
274	if (heartbeatTimeoutDeadlineNanos != 0) {	1✔
275	heartbeatTimeoutDeadlineNanos = computeHeartbeatTimeoutDeadlineNanos();	1✔
276	}
277	} catch (StatusRuntimeException e) {	1✔
278	// Not rethrowing to not fail activity implementation on intermittent connection or Temporal
279	// errors.
280	log.warn("Heartbeat failed", e);	1✔
281	nextHeartbeatDelay = HEARTBEAT_RETRY_WAIT_MILLIS;	1✔
282	} catch (Exception e) {	×
283	log.error("Unexpected exception", e);	×
UNCOV 284	nextHeartbeatDelay = HEARTBEAT_RETRY_WAIT_MILLIS;	×
285	}	1✔
286
287	scheduleNextHeartbeatLocked(nextHeartbeatDelay);	1✔
288	}	1✔
289
290	private void scheduleNextHeartbeatLocked(long delay) {
291	scheduledHeartbeat =	1✔
292	heartbeatExecutor.schedule(	1✔
293	() -> {
294	lock.lock();	1✔
295	try {
296	if (hasOutstandingHeartbeat) {	1✔
297	doHeartBeatLocked(lastDetails);	1✔
298	} else {
299	// if no new heartbeats have been submitted in the previous time interval, we
300	// don't need to throttle
301	// and the next heartbeat should go immediately without following a schedule.
302	scheduledHeartbeat = null;	1✔
303	}
304	} finally {
305	lock.unlock();	1✔
306	}
307	},	1✔
308	delay,
309	TimeUnit.MILLISECONDS);
310	}	1✔
311
312	private long computeHeartbeatTimeoutDeadlineNanos() {
313	return System.nanoTime()	1✔
314	+ TimeUnit.MILLISECONDS.toNanos(heartbeatTimeoutMillis + localHeartbeatTimeoutBufferMillis);	1✔
315	}
316
317	private void checkHeartbeatTimeoutDeadlineLocked() {
318	if (heartbeatTimedOut) {	1✔
319	throw new ActivityCanceledException(	1✔
320	info, new TimeoutFailure(null, null, TimeoutType.TIMEOUT_TYPE_HEARTBEAT));
321	}
322	if (heartbeatTimeoutDeadlineNanos != 0 && System.nanoTime() >= heartbeatTimeoutDeadlineNanos) {	1✔
323	heartbeatTimedOut = true;	1✔
324	log.warn(	1✔
325	"Activity heartbeat timed out locally. ActivityId={}, activityType={}",
326	info.getActivityId(),	1✔
327	info.getActivityType());	1✔
328	throw new ActivityCanceledException(	1✔
329	info, new TimeoutFailure(null, null, TimeoutType.TIMEOUT_TYPE_HEARTBEAT));
330	}
331	}	1✔
332
333	private void sendHeartbeatRequest(Object details) {
334	try {
335	RecordActivityTaskHeartbeatResponse status =	1✔
336	ActivityClientHelper.sendHeartbeatRequest(	1✔
337	service,
338	namespace,
339	identity,
340	info.getTaskToken(),	1✔
341	dataConverterWithActivityContext.toPayloads(details),	1✔
342	metricsScope);
343	if (status.getCancelRequested()) {	1✔
344	requestCancelLocked();	1✔
345	} else if (status.getActivityReset()) {	1!
UNCOV 346	lastException = new ActivityResetException(info);	×
347	} else if (status.getActivityPaused()) {	1!
UNCOV 348	lastException = new ActivityPausedException(info);	×
349	} else {
350	lastException = null;	1✔
351	}
352	} catch (StatusRuntimeException e) {	1✔
353	if (e.getStatus().getCode() == Status.Code.NOT_FOUND) {	1✔
354	lastException = new ActivityNotExistsException(info, e);	1✔
355	} else if (e.getStatus().getCode() == Status.Code.INVALID_ARGUMENT	1!
356	\|\| e.getStatus().getCode() == Status.Code.FAILED_PRECONDITION) {	1!
UNCOV 357	lastException = new ActivityCompletionFailureException(info, e);	×
358	} else {
359	throw e;	1✔
360	}
361	}	1✔
362	}	1✔
363
364	private void requestCancelLocked() {
365	ActivityCanceledException exception = new ActivityCanceledException(info);	1✔
366	lastException = exception;	1✔
367	cancellationSource.cancel(exception);	1✔
368	}	1✔
369
370	private static long getHeartbeatIntervalMs(
371	Duration activityHeartbeatTimeout,
372	Duration maxHeartbeatThrottleInterval,
373	Duration defaultHeartbeatThrottleInterval) {
374	long interval =
375	activityHeartbeatTimeout.isZero()	1✔
376	? defaultHeartbeatThrottleInterval.toMillis()	1✔
377	: (long) (0.8 * activityHeartbeatTimeout.toMillis());	1✔
378	return Math.min(interval, maxHeartbeatThrottleInterval.toMillis());	1✔
379	}
380	}

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