#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

91.06

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

package io.temporal.internal.worker;

import io.temporal.api.worker.v1.WorkerHeartbeat;
import io.temporal.api.workflowservice.v1.RecordWorkerHeartbeatRequest;
import io.temporal.serviceclient.WorkflowServiceStubs;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Supplier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Manages periodic worker heartbeat RPCs. Routes workers to per-namespace {@link
 * SharedNamespaceWorker} instances, each with its own scheduler.
 */
public class HeartbeatManager {
  private static final Logger log = LoggerFactory.getLogger(HeartbeatManager.class);

  private final WorkflowServiceStubs service;
  private final String identity;
  private final Duration interval;
  private final Map<String, SharedNamespaceWorker> namespaceWorkers = new HashMap<>();
  private final Set<String> unimplementedNamespaces = new HashSet<>();

  private final Object lock = new Object();

  public HeartbeatManager(WorkflowServiceStubs service, String identity, Duration interval) {
    this.service = service;
    this.identity = identity;
    this.interval = interval;
  }

  /**
   * Register a worker's heartbeat callback. Creates a per-namespace SharedNamespaceWorker if this
   * is the first worker for the given namespace.
   */
  public void registerWorker(
      String namespace, String workerInstanceKey, Supplier<WorkerHeartbeat> callback) {
    synchronized (lock) {
      if (unimplementedNamespaces.contains(namespace)) {
        return;
      }
      namespaceWorkers.compute(
          namespace,
          (ns, existing) -> {
            if (existing != null && !existing.isShutdown()) {
              existing.registerWorker(workerInstanceKey, callback);
              return existing;
            }
            SharedNamespaceWorker nsWorker =
                new SharedNamespaceWorker(this, service, ns, identity, interval);
            nsWorker.registerWorker(workerInstanceKey, callback);
            return nsWorker;
          });
    }
  }

  /** Unregister a worker. Stops the namespace worker if no workers remain for that namespace. */
  public void unregisterWorker(String namespace, String workerInstanceKey) {
    synchronized (lock) {
      SharedNamespaceWorker nsWorker = namespaceWorkers.get(namespace);
      if (nsWorker == null) {
        return;
      }
      nsWorker.unregisterWorker(workerInstanceKey);
      if (nsWorker.isEmpty()) {
        nsWorker.shutdown();
        namespaceWorkers.remove(namespace);
      }
    }
  }

  public void shutdown() {
    synchronized (lock) {
      for (SharedNamespaceWorker nsWorker : namespaceWorkers.values()) {
        nsWorker.shutdown();
      }
      namespaceWorkers.clear();
    }
  }

  /**
   * Called from the scheduler thread when the server returns UNIMPLEMENTED. Uses
   * scheduler.shutdown() (graceful) instead of shutdownNow() to avoid interrupting the
   * currently-executing tick, and skips awaitTermination since we're on the scheduler thread
   * itself.
   */
  void markNamespaceUnimplemented(String namespace) {
    synchronized (lock) {
      unimplementedNamespaces.add(namespace);
      SharedNamespaceWorker nsWorker = namespaceWorkers.remove(namespace);
      if (nsWorker != null) {
        nsWorker.stopScheduling();
      }
    }
  }

  /**
   * Handles heartbeating for all workers in a specific namespace. Each instance owns its own
   * scheduler thread and callback map.
   */
  static class SharedNamespaceWorker {
    private final HeartbeatManager manager;
    private final WorkflowServiceStubs service;
    private final String namespace;
    private final String identity;
    private final ConcurrentHashMap<String, Supplier<WorkerHeartbeat>> callbacks =
        new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler;

    SharedNamespaceWorker(
        HeartbeatManager manager,
        WorkflowServiceStubs service,
        String namespace,
        String identity,
        Duration interval) {
      this.manager = manager;
      this.service = service;
      this.namespace = namespace;
      this.identity = identity;
      this.scheduler =
          Executors.newSingleThreadScheduledExecutor(
              r -> {
                Thread t = new Thread(r, "worker-heartbeat-" + namespace);
                t.setDaemon(true);
                return t;
              });
      scheduler.scheduleAtFixedRate(
          this::heartbeatTick, 0, interval.toMillis(), TimeUnit.MILLISECONDS);
    }

    void registerWorker(String workerInstanceKey, Supplier<WorkerHeartbeat> callback) {
      callbacks.put(workerInstanceKey, callback);
    }

    void unregisterWorker(String workerInstanceKey) {
      callbacks.remove(workerInstanceKey);
    }

    boolean isEmpty() {
      return callbacks.isEmpty();
    }

    boolean isShutdown() {
      return scheduler.isShutdown();
    }

    /** Stops scheduling new ticks. Safe to call from the scheduler thread itself. */
    void stopScheduling() {
      scheduler.shutdown();
    }

    /** Full shutdown from an external thread. Interrupts in-flight work and waits. */
    void shutdown() {
      scheduler.shutdownNow();
      try {
        scheduler.awaitTermination(5, TimeUnit.SECONDS);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    }

    private void heartbeatTick() {
      if (callbacks.isEmpty()) return;

      List<WorkerHeartbeat> heartbeats = new ArrayList<>();
      for (Map.Entry<String, Supplier<WorkerHeartbeat>> entry : callbacks.entrySet()) {
        try {
          heartbeats.add(entry.getValue().get());
        } catch (Exception e) {
          log.warn(
              "Failed to build heartbeat for worker {} in namespace {}",
              entry.getKey(),
              namespace,
              e);
        }
      }

      if (heartbeats.isEmpty()) return;

      try {
        service
            .blockingStub()
            .recordWorkerHeartbeat(
                RecordWorkerHeartbeatRequest.newBuilder()
                    .setNamespace(namespace)
                    .setIdentity(identity)
                    .addAllWorkerHeartbeat(heartbeats)
                    .build());
      } catch (io.grpc.StatusRuntimeException e) {
        if (e.getStatus().getCode() == io.grpc.Status.Code.UNIMPLEMENTED) {
          log.warn(
              "Server does not support worker heartbeats for namespace {}, disabling", namespace);
          manager.markNamespaceUnimplemented(namespace);
          return;
        }
        log.warn("Failed to send worker heartbeat for namespace {}", namespace, e);
      } catch (Exception e) {
        log.warn("Failed to send worker heartbeat for namespace {}", namespace, e);
      }
    }
  }
}

1	package io.temporal.internal.worker;
2
3	import io.temporal.api.worker.v1.WorkerHeartbeat;
4	import io.temporal.api.workflowservice.v1.RecordWorkerHeartbeatRequest;
5	import io.temporal.serviceclient.WorkflowServiceStubs;
6	import java.time.Duration;
7	import java.util.*;
8	import java.util.concurrent.*;
9	import java.util.function.Supplier;
10	import org.slf4j.Logger;
11	import org.slf4j.LoggerFactory;
12
13	/**
14	* Manages periodic worker heartbeat RPCs. Routes workers to per-namespace {@link
15	* SharedNamespaceWorker} instances, each with its own scheduler.
16	*/
17	public class HeartbeatManager {
18	private static final Logger log = LoggerFactory.getLogger(HeartbeatManager.class);	1✔
19
20	private final WorkflowServiceStubs service;
21	private final String identity;
22	private final Duration interval;
23	private final Map<String, SharedNamespaceWorker> namespaceWorkers = new HashMap<>();	1✔
24	private final Set<String> unimplementedNamespaces = new HashSet<>();	1✔
25
26	private final Object lock = new Object();	1✔
27
28	public HeartbeatManager(WorkflowServiceStubs service, String identity, Duration interval) {	1✔
29	this.service = service;	1✔
30	this.identity = identity;	1✔
31	this.interval = interval;	1✔
32	}	1✔
33
34	/**
35	* Register a worker's heartbeat callback. Creates a per-namespace SharedNamespaceWorker if this
36	* is the first worker for the given namespace.
37	*/
38	public void registerWorker(
39	String namespace, String workerInstanceKey, Supplier<WorkerHeartbeat> callback) {
40	synchronized (lock) {	1✔
41	if (unimplementedNamespaces.contains(namespace)) {	1!
42	return;	×
43	}
44	namespaceWorkers.compute(	1✔
45	namespace,
46	(ns, existing) -> {
47	if (existing != null && !existing.isShutdown()) {	1!
48	existing.registerWorker(workerInstanceKey, callback);	1✔
49	return existing;	1✔
50	}
51	SharedNamespaceWorker nsWorker =	1✔
52	new SharedNamespaceWorker(this, service, ns, identity, interval);
53	nsWorker.registerWorker(workerInstanceKey, callback);	1✔
54	return nsWorker;	1✔
55	});
56	}	1✔
57	}	1✔
58
59	/** Unregister a worker. Stops the namespace worker if no workers remain for that namespace. */
60	public void unregisterWorker(String namespace, String workerInstanceKey) {
61	synchronized (lock) {	1✔
62	SharedNamespaceWorker nsWorker = namespaceWorkers.get(namespace);	1✔
63	if (nsWorker == null) {	1✔
64	return;	1✔
65	}
66	nsWorker.unregisterWorker(workerInstanceKey);	1✔
67	if (nsWorker.isEmpty()) {	1✔
68	nsWorker.shutdown();	1✔
69	namespaceWorkers.remove(namespace);	1✔
70	}
71	}	1✔
72	}	1✔
73
74	public void shutdown() {
75	synchronized (lock) {	1✔
76	for (SharedNamespaceWorker nsWorker : namespaceWorkers.values()) {	1✔
77	nsWorker.shutdown();	1✔
78	}	1✔
79	namespaceWorkers.clear();	1✔
80	}	1✔
81	}	1✔
82
83	/**
84	* Called from the scheduler thread when the server returns UNIMPLEMENTED. Uses
85	* scheduler.shutdown() (graceful) instead of shutdownNow() to avoid interrupting the
86	* currently-executing tick, and skips awaitTermination since we're on the scheduler thread
87	* itself.
88	*/
89	void markNamespaceUnimplemented(String namespace) {
90	synchronized (lock) {	1✔
91	unimplementedNamespaces.add(namespace);	1✔
92	SharedNamespaceWorker nsWorker = namespaceWorkers.remove(namespace);	1✔
93	if (nsWorker != null) {	1✔
94	nsWorker.stopScheduling();	1✔
95	}
96	}	1✔
97	}	1✔
98
99	/**
100	* Handles heartbeating for all workers in a specific namespace. Each instance owns its own
101	* scheduler thread and callback map.
102	*/
103	static class SharedNamespaceWorker {
104	private final HeartbeatManager manager;
105	private final WorkflowServiceStubs service;
106	private final String namespace;
107	private final String identity;
108	private final ConcurrentHashMap<String, Supplier<WorkerHeartbeat>> callbacks =	1✔
109	new ConcurrentHashMap<>();
110	private final ScheduledExecutorService scheduler;
111
112	SharedNamespaceWorker(
113	HeartbeatManager manager,
114	WorkflowServiceStubs service,
115	String namespace,
116	String identity,
117	Duration interval) {	1✔
118	this.manager = manager;	1✔
119	this.service = service;	1✔
120	this.namespace = namespace;	1✔
121	this.identity = identity;	1✔
122	this.scheduler =	1✔
123	Executors.newSingleThreadScheduledExecutor(	1✔
124	r -> {
125	Thread t = new Thread(r, "worker-heartbeat-" + namespace);	1✔
126	t.setDaemon(true);	1✔
127	return t;	1✔
128	});
129	scheduler.scheduleAtFixedRate(	1✔
130	this::heartbeatTick, 0, interval.toMillis(), TimeUnit.MILLISECONDS);	1✔
131	}	1✔
132
133	void registerWorker(String workerInstanceKey, Supplier<WorkerHeartbeat> callback) {
134	callbacks.put(workerInstanceKey, callback);	1✔
135	}	1✔
136
137	void unregisterWorker(String workerInstanceKey) {
138	callbacks.remove(workerInstanceKey);	1✔
139	}	1✔
140
141	boolean isEmpty() {
142	return callbacks.isEmpty();	1✔
143	}
144
145	boolean isShutdown() {
146	return scheduler.isShutdown();	1✔
147	}
148
149	/** Stops scheduling new ticks. Safe to call from the scheduler thread itself. */
150	void stopScheduling() {
151	scheduler.shutdown();	1✔
152	}	1✔
153
154	/** Full shutdown from an external thread. Interrupts in-flight work and waits. */
155	void shutdown() {
156	scheduler.shutdownNow();	1✔
157	try {
158	scheduler.awaitTermination(5, TimeUnit.SECONDS);	1✔
159	} catch (InterruptedException e) {	×
160	Thread.currentThread().interrupt();	×
161	}	1✔
162	}	1✔
163
164	private void heartbeatTick() {
165	if (callbacks.isEmpty()) return;	1✔
166
167	List<WorkerHeartbeat> heartbeats = new ArrayList<>();	1✔
168	for (Map.Entry<String, Supplier<WorkerHeartbeat>> entry : callbacks.entrySet()) {	1✔
169	try {
170	heartbeats.add(entry.getValue().get());	1✔
171	} catch (Exception e) {	×
172	log.warn(	×
173	"Failed to build heartbeat for worker {} in namespace {}",
174	entry.getKey(),	×
175	namespace,
176	e);
177	}	1✔
178	}	1✔
179
180	if (heartbeats.isEmpty()) return;	1!
181
182	try {
183	service	1✔
184	.blockingStub()	1✔
185	.recordWorkerHeartbeat(	1✔
186	RecordWorkerHeartbeatRequest.newBuilder()	1✔
187	.setNamespace(namespace)	1✔
188	.setIdentity(identity)	1✔
189	.addAllWorkerHeartbeat(heartbeats)	1✔
190	.build());	1✔
191	} catch (io.grpc.StatusRuntimeException e) {	1✔
192	if (e.getStatus().getCode() == io.grpc.Status.Code.UNIMPLEMENTED) {	1!
193	log.warn(	1✔
194	"Server does not support worker heartbeats for namespace {}, disabling", namespace);
195	manager.markNamespaceUnimplemented(namespace);	1✔
196	return;	1✔
197	}
UNCOV 198	log.warn("Failed to send worker heartbeat for namespace {}", namespace, e);	×
199	} catch (Exception e) {	1✔
200	log.warn("Failed to send worker heartbeat for namespace {}", namespace, e);	1✔
201	}	1✔
202	}	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