#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

84.28

/temporal-sdk/src/main/java/io/temporal/worker/tuning/ResourceBasedSlotSupplier.java

package io.temporal.worker.tuning;

import java.time.Duration;
import java.time.Instant;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;

/** Implements a {@link SlotSupplier} based on resource usage for a particular slot type. */
public class ResourceBasedSlotSupplier<SI extends SlotInfo> implements SlotSupplier<SI> {

  private final ResourceBasedController resourceController;
  private final ResourceBasedSlotOptions options;
  private Instant lastSlotIssuedAt = Instant.EPOCH;
  // For slot reservations that are waiting to re-check resource usage
  private final ScheduledExecutorService scheduler;
  private static ScheduledExecutorService defaultScheduler;

  /**
   * Construct a slot supplier for workflow tasks with the given resource controller and options.
   *
   * <p>The resource controller must be the same among all slot suppliers in a worker. If you want
   * to use resource-based tuning for all slot suppliers, prefer {@link ResourceBasedTuner}.
   */
  public static ResourceBasedSlotSupplier<WorkflowSlotInfo> createForWorkflow(
      ResourceBasedController resourceBasedController, ResourceBasedSlotOptions options) {
    return new ResourceBasedSlotSupplier<>(
        WorkflowSlotInfo.class, resourceBasedController, options, null);
  }

  /**
   * As {@link #createForWorkflow(ResourceBasedController, ResourceBasedSlotOptions)}, but allows
   * overriding the internal thread pool. It is recommended to share the same executor across all
   * resource based slot suppliers in a worker.
   */
  public static ResourceBasedSlotSupplier<WorkflowSlotInfo> createForWorkflow(
      ResourceBasedController resourceBasedController,
      ResourceBasedSlotOptions options,
      ScheduledExecutorService scheduler) {
    return new ResourceBasedSlotSupplier<>(
        WorkflowSlotInfo.class, resourceBasedController, options, scheduler);
  }

  /**
   * Construct a slot supplier for activity tasks with the given resource controller and options.
   *
   * <p>The resource controller must be the same among all slot suppliers in a worker. If you want
   * to use resource-based tuning for all slot suppliers, prefer {@link ResourceBasedTuner}.
   */
  public static ResourceBasedSlotSupplier<ActivitySlotInfo> createForActivity(
      ResourceBasedController resourceBasedController, ResourceBasedSlotOptions options) {
    return new ResourceBasedSlotSupplier<>(
        ActivitySlotInfo.class, resourceBasedController, options, null);
  }

  /**
   * As {@link #createForActivity(ResourceBasedController, ResourceBasedSlotOptions)}, but allows
   * overriding the internal thread pool. It is recommended to share the same executor across all
   * resource based slot suppliers in a worker.
   */
  public static ResourceBasedSlotSupplier<ActivitySlotInfo> createForActivity(
      ResourceBasedController resourceBasedController,
      ResourceBasedSlotOptions options,
      ScheduledExecutorService scheduler) {
    return new ResourceBasedSlotSupplier<>(
        ActivitySlotInfo.class, resourceBasedController, options, scheduler);
  }

  /**
   * Construct a slot supplier for local activities with the given resource controller and options.
   *
   * <p>The resource controller must be the same among all slot suppliers in a worker. If you want
   * to use resource-based tuning for all slot suppliers, prefer {@link ResourceBasedTuner}.
   */
  public static ResourceBasedSlotSupplier<LocalActivitySlotInfo> createForLocalActivity(
      ResourceBasedController resourceBasedController, ResourceBasedSlotOptions options) {
    return new ResourceBasedSlotSupplier<>(
        LocalActivitySlotInfo.class, resourceBasedController, options, null);
  }

  /**
   * As {@link #createForLocalActivity(ResourceBasedController, ResourceBasedSlotOptions)}, but
   * allows overriding the internal thread pool. It is recommended to share the same executor across
   * all resource based slot suppliers in a worker.
   */
  public static ResourceBasedSlotSupplier<LocalActivitySlotInfo> createForLocalActivity(
      ResourceBasedController resourceBasedController,
      ResourceBasedSlotOptions options,
      ScheduledExecutorService scheduler) {
    return new ResourceBasedSlotSupplier<>(
        LocalActivitySlotInfo.class, resourceBasedController, options, scheduler);
  }

  /**
   * Construct a slot supplier for nexus tasks with the given resource controller and options.
   *
   * <p>The resource controller must be the same among all slot suppliers in a worker. If you want
   * to use resource-based tuning for all slot suppliers, prefer {@link ResourceBasedTuner}.
   */
  public static ResourceBasedSlotSupplier<NexusSlotInfo> createForNexus(
      ResourceBasedController resourceBasedController, ResourceBasedSlotOptions options) {
    return new ResourceBasedSlotSupplier<>(
        NexusSlotInfo.class, resourceBasedController, options, null);
  }

  /**
   * As {@link #createForNexus(ResourceBasedController, ResourceBasedSlotOptions)}, but allows
   * overriding the internal thread pool. It is recommended to share the same executor across all
   * resource based slot suppliers in a worker.
   */
  public static ResourceBasedSlotSupplier<NexusSlotInfo> createForNexus(
      ResourceBasedController resourceBasedController,
      ResourceBasedSlotOptions options,
      ScheduledExecutorService scheduler) {
    return new ResourceBasedSlotSupplier<>(
        NexusSlotInfo.class, resourceBasedController, options, scheduler);
  }

  private ResourceBasedSlotSupplier(
      Class<SI> clazz,
      ResourceBasedController resourceBasedController,
      ResourceBasedSlotOptions options,
      ScheduledExecutorService scheduler) {
    this.resourceController = resourceBasedController;
    if (scheduler == null) {
      this.scheduler = getDefaultScheduler();
    } else {
      this.scheduler = scheduler;
    }
    // Merge default options for any unset fields
    if (WorkflowSlotInfo.class.isAssignableFrom(clazz)) {
      this.options =
          ResourceBasedSlotOptions.newBuilder()
              .setMinimumSlots(
                  options.getMinimumSlots() == 0
                      ? ResourceBasedTuner.DEFAULT_WORKFLOW_SLOT_OPTIONS.getMinimumSlots()
                      : options.getMinimumSlots())
              .setMaximumSlots(
                  options.getMaximumSlots() == 0
                      ? ResourceBasedTuner.DEFAULT_WORKFLOW_SLOT_OPTIONS.getMaximumSlots()
                      : options.getMaximumSlots())
              .setRampThrottle(
                  options.getRampThrottle() == null
                      ? ResourceBasedTuner.DEFAULT_WORKFLOW_SLOT_OPTIONS.getRampThrottle()
                      : options.getRampThrottle())
              .build();
    } else if (ActivitySlotInfo.class.isAssignableFrom(clazz)
        || LocalActivitySlotInfo.class.isAssignableFrom(clazz)) {
      this.options =
          ResourceBasedSlotOptions.newBuilder()
              .setMinimumSlots(
                  options.getMinimumSlots() == 0
                      ? ResourceBasedTuner.DEFAULT_ACTIVITY_SLOT_OPTIONS.getMinimumSlots()
                      : options.getMinimumSlots())
              .setMaximumSlots(
                  options.getMaximumSlots() == 0
                      ? ResourceBasedTuner.DEFAULT_ACTIVITY_SLOT_OPTIONS.getMaximumSlots()
                      : options.getMaximumSlots())
              .setRampThrottle(
                  options.getRampThrottle() == null
                      ? ResourceBasedTuner.DEFAULT_ACTIVITY_SLOT_OPTIONS.getRampThrottle()
                      : options.getRampThrottle())
              .build();
    } else {
      this.options =
          ResourceBasedSlotOptions.newBuilder()
              .setMinimumSlots(
                  options.getMinimumSlots() == 0
                      ? ResourceBasedTuner.DEFAULT_NEXUS_SLOT_OPTIONS.getMinimumSlots()
                      : options.getMinimumSlots())
              .setMaximumSlots(
                  options.getMaximumSlots() == 0
                      ? ResourceBasedTuner.DEFAULT_NEXUS_SLOT_OPTIONS.getMaximumSlots()
                      : options.getMaximumSlots())
              .setRampThrottle(
                  options.getRampThrottle() == null
                      ? ResourceBasedTuner.DEFAULT_NEXUS_SLOT_OPTIONS.getRampThrottle()
                      : options.getRampThrottle())
              .build();
    }
  }

  @Override
  public SlotSupplierFuture reserveSlot(SlotReserveContext<SI> ctx) throws Exception {
    if (ctx.getNumIssuedSlots() < options.getMinimumSlots()) {
      return SlotSupplierFuture.completedFuture(new SlotPermit());
    }
    return tryReserveSlot(ctx)
        .map(SlotSupplierFuture::completedFuture)
        .orElseGet(() -> scheduleSlotAcquisition(ctx));
  }

  private SlotSupplierFuture scheduleSlotAcquisition(SlotReserveContext<SI> ctx) {
    CompletableFuture<SlotPermit> resultFuture = new CompletableFuture<>();
    AtomicReference<ScheduledFuture<?>> taskRef = new AtomicReference<>();

    Runnable pollingTask =
        new Runnable() {
          @Override
          public void run() {
            if (resultFuture.isDone()) {
              return; // Already completed or cancelled
            }

            try {
              Optional<SlotPermit> permit = tryReserveSlot(ctx);
              if (permit.isPresent()) {
                resultFuture.complete(permit.get());
              } else {
                taskRef.set(scheduler.schedule(this, 10, TimeUnit.MILLISECONDS));
              }
            } catch (Exception e) {
              resultFuture.completeExceptionally(e);
            }
          }
        };

    // Calculate initial delay based on ramp throttle
    Duration mustWaitFor;
    try {
      mustWaitFor = options.getRampThrottle().minus(timeSinceLastSlotIssued());
    } catch (ArithmeticException e) {
      mustWaitFor = Duration.ZERO;
    }

    long initialDelayMs = Math.max(0, mustWaitFor.toMillis());

    // Schedule the initial attempt
    taskRef.set(scheduler.schedule(pollingTask, initialDelayMs, TimeUnit.MILLISECONDS));

    return SlotSupplierFuture.fromCompletableFuture(
        resultFuture,
        () -> {
          // Cancel the scheduled task when aborting
          ScheduledFuture<?> task = taskRef.get();
          if (task != null) {
            task.cancel(true);
          }
        });
  }

  @Override
  public Optional<SlotPermit> tryReserveSlot(SlotReserveContext<SI> ctx) {
    int numIssued = ctx.getNumIssuedSlots();
    if (numIssued < options.getMinimumSlots()
        || (timeSinceLastSlotIssued().compareTo(options.getRampThrottle()) > 0
            && numIssued < options.getMaximumSlots()
            && resourceController.pidDecision())) {
      lastSlotIssuedAt = Instant.now();
      return Optional.of(new SlotPermit());
    }
    return Optional.empty();
  }

  @Override
  public void markSlotUsed(SlotMarkUsedContext<SI> ctx) {}

  @Override
  public void releaseSlot(SlotReleaseContext<SI> ctx) {}

  public ResourceBasedController getResourceController() {
    return resourceController;
  }

  private Duration timeSinceLastSlotIssued() {
    return Duration.between(lastSlotIssuedAt, Instant.now());
  }

  // Polyfill for Java 9 delayedExecutor
  private Executor delayedExecutor(long delay) {
    return r -> scheduler.schedule(() -> scheduler.execute(r), delay, TimeUnit.MILLISECONDS);
  }

  private static ScheduledExecutorService getDefaultScheduler() {
    synchronized (ResourceBasedSlotSupplier.class) {
      if (defaultScheduler == null) {
        defaultScheduler =
            Executors.newScheduledThreadPool(
                // Two threads seem needed here, so that reading PID decisions doesn't interfere
                // overly with firing off scheduled tasks or one another.
                2,
                r -> {
                  Thread t = new Thread(r);
                  t.setName("ResourceBasedSlotSupplier.scheduler");
                  t.setDaemon(true);
                  return t;
                });
      }
      return defaultScheduler;
    }
  }
}

1	package io.temporal.worker.tuning;
2
3	import java.time.Duration;
4	import java.time.Instant;
5	import java.util.Optional;
6	import java.util.concurrent.*;
7	import java.util.concurrent.atomic.AtomicReference;
8
9	/** Implements a {@link SlotSupplier} based on resource usage for a particular slot type. */
10	public class ResourceBasedSlotSupplier<SI extends SlotInfo> implements SlotSupplier<SI> {
11
12	private final ResourceBasedController resourceController;
13	private final ResourceBasedSlotOptions options;
14	private Instant lastSlotIssuedAt = Instant.EPOCH;	1✔
15	// For slot reservations that are waiting to re-check resource usage
16	private final ScheduledExecutorService scheduler;
17	private static ScheduledExecutorService defaultScheduler;
18
19	/**
20	* Construct a slot supplier for workflow tasks with the given resource controller and options.
21	*
22	* <p>The resource controller must be the same among all slot suppliers in a worker. If you want
23	* to use resource-based tuning for all slot suppliers, prefer {@link ResourceBasedTuner}.
24	*/
25	public static ResourceBasedSlotSupplier<WorkflowSlotInfo> createForWorkflow(
26	ResourceBasedController resourceBasedController, ResourceBasedSlotOptions options) {
27	return new ResourceBasedSlotSupplier<>(	×
28	WorkflowSlotInfo.class, resourceBasedController, options, null);
29	}
30
31	/**
32	* As {@link #createForWorkflow(ResourceBasedController, ResourceBasedSlotOptions)}, but allows
33	* overriding the internal thread pool. It is recommended to share the same executor across all
34	* resource based slot suppliers in a worker.
35	*/
36	public static ResourceBasedSlotSupplier<WorkflowSlotInfo> createForWorkflow(
37	ResourceBasedController resourceBasedController,
38	ResourceBasedSlotOptions options,
39	ScheduledExecutorService scheduler) {
40	return new ResourceBasedSlotSupplier<>(	1✔
41	WorkflowSlotInfo.class, resourceBasedController, options, scheduler);
42	}
43
44	/**
45	* Construct a slot supplier for activity tasks with the given resource controller and options.
46	*
47	* <p>The resource controller must be the same among all slot suppliers in a worker. If you want
48	* to use resource-based tuning for all slot suppliers, prefer {@link ResourceBasedTuner}.
49	*/
50	public static ResourceBasedSlotSupplier<ActivitySlotInfo> createForActivity(
51	ResourceBasedController resourceBasedController, ResourceBasedSlotOptions options) {
52	return new ResourceBasedSlotSupplier<>(	1✔
53	ActivitySlotInfo.class, resourceBasedController, options, null);
54	}
55
56	/**
57	* As {@link #createForActivity(ResourceBasedController, ResourceBasedSlotOptions)}, but allows
58	* overriding the internal thread pool. It is recommended to share the same executor across all
59	* resource based slot suppliers in a worker.
60	*/
61	public static ResourceBasedSlotSupplier<ActivitySlotInfo> createForActivity(
62	ResourceBasedController resourceBasedController,
63	ResourceBasedSlotOptions options,
64	ScheduledExecutorService scheduler) {
65	return new ResourceBasedSlotSupplier<>(	1✔
66	ActivitySlotInfo.class, resourceBasedController, options, scheduler);
67	}
68
69	/**
70	* Construct a slot supplier for local activities with the given resource controller and options.
71	*
72	* <p>The resource controller must be the same among all slot suppliers in a worker. If you want
73	* to use resource-based tuning for all slot suppliers, prefer {@link ResourceBasedTuner}.
74	*/
75	public static ResourceBasedSlotSupplier<LocalActivitySlotInfo> createForLocalActivity(
76	ResourceBasedController resourceBasedController, ResourceBasedSlotOptions options) {
77	return new ResourceBasedSlotSupplier<>(	1✔
78	LocalActivitySlotInfo.class, resourceBasedController, options, null);
79	}
80
81	/**
82	* As {@link #createForLocalActivity(ResourceBasedController, ResourceBasedSlotOptions)}, but
83	* allows overriding the internal thread pool. It is recommended to share the same executor across
84	* all resource based slot suppliers in a worker.
85	*/
86	public static ResourceBasedSlotSupplier<LocalActivitySlotInfo> createForLocalActivity(
87	ResourceBasedController resourceBasedController,
88	ResourceBasedSlotOptions options,
89	ScheduledExecutorService scheduler) {
90	return new ResourceBasedSlotSupplier<>(	1✔
91	LocalActivitySlotInfo.class, resourceBasedController, options, scheduler);
92	}
93
94	/**
95	* Construct a slot supplier for nexus tasks with the given resource controller and options.
96	*
97	* <p>The resource controller must be the same among all slot suppliers in a worker. If you want
98	* to use resource-based tuning for all slot suppliers, prefer {@link ResourceBasedTuner}.
99	*/
100	public static ResourceBasedSlotSupplier<NexusSlotInfo> createForNexus(
101	ResourceBasedController resourceBasedController, ResourceBasedSlotOptions options) {
102	return new ResourceBasedSlotSupplier<>(	×
103	NexusSlotInfo.class, resourceBasedController, options, null);
104	}
105
106	/**
107	* As {@link #createForNexus(ResourceBasedController, ResourceBasedSlotOptions)}, but allows
108	* overriding the internal thread pool. It is recommended to share the same executor across all
109	* resource based slot suppliers in a worker.
110	*/
111	public static ResourceBasedSlotSupplier<NexusSlotInfo> createForNexus(
112	ResourceBasedController resourceBasedController,
113	ResourceBasedSlotOptions options,
114	ScheduledExecutorService scheduler) {
115	return new ResourceBasedSlotSupplier<>(	1✔
116	NexusSlotInfo.class, resourceBasedController, options, scheduler);
117	}
118
119	private ResourceBasedSlotSupplier(
120	Class<SI> clazz,
121	ResourceBasedController resourceBasedController,
122	ResourceBasedSlotOptions options,
123	ScheduledExecutorService scheduler) {	1✔
124	this.resourceController = resourceBasedController;	1✔
125	if (scheduler == null) {	1✔
126	this.scheduler = getDefaultScheduler();	1✔
127	} else {
128	this.scheduler = scheduler;	1✔
129	}
130	// Merge default options for any unset fields
131	if (WorkflowSlotInfo.class.isAssignableFrom(clazz)) {	1✔
132	this.options =	1✔
133	ResourceBasedSlotOptions.newBuilder()	1✔
134	.setMinimumSlots(	1✔
135	options.getMinimumSlots() == 0	1!
136	? ResourceBasedTuner.DEFAULT_WORKFLOW_SLOT_OPTIONS.getMinimumSlots()	×
137	: options.getMinimumSlots())	1✔
138	.setMaximumSlots(	1✔
139	options.getMaximumSlots() == 0	1!
140	? ResourceBasedTuner.DEFAULT_WORKFLOW_SLOT_OPTIONS.getMaximumSlots()	×
141	: options.getMaximumSlots())	1✔
142	.setRampThrottle(	1✔
143	options.getRampThrottle() == null	1!
144	? ResourceBasedTuner.DEFAULT_WORKFLOW_SLOT_OPTIONS.getRampThrottle()	×
145	: options.getRampThrottle())	1✔
146	.build();	1✔
147	} else if (ActivitySlotInfo.class.isAssignableFrom(clazz)	1✔
148	\|\| LocalActivitySlotInfo.class.isAssignableFrom(clazz)) {	1✔
149	this.options =	1✔
150	ResourceBasedSlotOptions.newBuilder()	1✔
151	.setMinimumSlots(	1✔
152	options.getMinimumSlots() == 0	1✔
153	? ResourceBasedTuner.DEFAULT_ACTIVITY_SLOT_OPTIONS.getMinimumSlots()	1✔
154	: options.getMinimumSlots())	1✔
155	.setMaximumSlots(	1✔
156	options.getMaximumSlots() == 0	1!
157	? ResourceBasedTuner.DEFAULT_ACTIVITY_SLOT_OPTIONS.getMaximumSlots()	×
158	: options.getMaximumSlots())	1✔
159	.setRampThrottle(	1✔
160	options.getRampThrottle() == null	1✔
161	? ResourceBasedTuner.DEFAULT_ACTIVITY_SLOT_OPTIONS.getRampThrottle()	1✔
162	: options.getRampThrottle())	1✔
163	.build();	1✔
164	} else {
165	this.options =	1✔
166	ResourceBasedSlotOptions.newBuilder()	1✔
167	.setMinimumSlots(	1✔
168	options.getMinimumSlots() == 0	1!
169	? ResourceBasedTuner.DEFAULT_NEXUS_SLOT_OPTIONS.getMinimumSlots()	×
170	: options.getMinimumSlots())	1✔
171	.setMaximumSlots(	1✔
172	options.getMaximumSlots() == 0	1!
173	? ResourceBasedTuner.DEFAULT_NEXUS_SLOT_OPTIONS.getMaximumSlots()	×
174	: options.getMaximumSlots())	1✔
175	.setRampThrottle(	1✔
176	options.getRampThrottle() == null	1!
177	? ResourceBasedTuner.DEFAULT_NEXUS_SLOT_OPTIONS.getRampThrottle()	×
178	: options.getRampThrottle())	1✔
179	.build();	1✔
180	}
181	}	1✔
182
183	@Override
184	public SlotSupplierFuture reserveSlot(SlotReserveContext<SI> ctx) throws Exception {
185	if (ctx.getNumIssuedSlots() < options.getMinimumSlots()) {	1✔
186	return SlotSupplierFuture.completedFuture(new SlotPermit());	1✔
187	}
188	return tryReserveSlot(ctx)	1✔
189	.map(SlotSupplierFuture::completedFuture)	1✔
190	.orElseGet(() -> scheduleSlotAcquisition(ctx));	1✔
191	}
192
193	private SlotSupplierFuture scheduleSlotAcquisition(SlotReserveContext<SI> ctx) {
194	CompletableFuture<SlotPermit> resultFuture = new CompletableFuture<>();	1✔
195	AtomicReference<ScheduledFuture<?>> taskRef = new AtomicReference<>();	1✔
196
197	Runnable pollingTask =	1✔
198	new Runnable() {	1✔
199	@Override
200	public void run() {
201	if (resultFuture.isDone()) {	1!
UNCOV 202	return; // Already completed or cancelled	×
203	}
204
205	try {
206	Optional<SlotPermit> permit = tryReserveSlot(ctx);	1✔
207	if (permit.isPresent()) {	1✔
208	resultFuture.complete(permit.get());	1✔
209	} else {
210	taskRef.set(scheduler.schedule(this, 10, TimeUnit.MILLISECONDS));	1✔
211	}
212	} catch (Exception e) {	×
213	resultFuture.completeExceptionally(e);	×
214	}	1✔
215	}	1✔
216	};
217
218	// Calculate initial delay based on ramp throttle
219	Duration mustWaitFor;
220	try {
221	mustWaitFor = options.getRampThrottle().minus(timeSinceLastSlotIssued());	1✔
222	} catch (ArithmeticException e) {	×
223	mustWaitFor = Duration.ZERO;	×
224	}	1✔
225
226	long initialDelayMs = Math.max(0, mustWaitFor.toMillis());	1✔
227
228	// Schedule the initial attempt
229	taskRef.set(scheduler.schedule(pollingTask, initialDelayMs, TimeUnit.MILLISECONDS));	1✔
230
231	return SlotSupplierFuture.fromCompletableFuture(	1✔
232	resultFuture,
233	() -> {
234	// Cancel the scheduled task when aborting
235	ScheduledFuture<?> task = taskRef.get();	1✔
236	if (task != null) {	1!
237	task.cancel(true);	1✔
238	}
239	});	1✔
240	}
241
242	@Override
243	public Optional<SlotPermit> tryReserveSlot(SlotReserveContext<SI> ctx) {
244	int numIssued = ctx.getNumIssuedSlots();	1✔
245	if (numIssued < options.getMinimumSlots()	1✔
246	\|\| (timeSinceLastSlotIssued().compareTo(options.getRampThrottle()) > 0	1✔
247	&& numIssued < options.getMaximumSlots()	1!
248	&& resourceController.pidDecision())) {	1✔
249	lastSlotIssuedAt = Instant.now();	1✔
250	return Optional.of(new SlotPermit());	1✔
251	}
252	return Optional.empty();	1✔
253	}
254
255	@Override
256	public void markSlotUsed(SlotMarkUsedContext<SI> ctx) {}	1✔
257
258	@Override
259	public void releaseSlot(SlotReleaseContext<SI> ctx) {}	1✔
260
261	public ResourceBasedController getResourceController() {
262	return resourceController;	1✔
263	}
264
265	private Duration timeSinceLastSlotIssued() {
266	return Duration.between(lastSlotIssuedAt, Instant.now());	1✔
267	}
268
269	// Polyfill for Java 9 delayedExecutor
270	private Executor delayedExecutor(long delay) {
271	return r -> scheduler.schedule(() -> scheduler.execute(r), delay, TimeUnit.MILLISECONDS);	×
272	}
273
274	private static ScheduledExecutorService getDefaultScheduler() {
275	synchronized (ResourceBasedSlotSupplier.class) {	1✔
276	if (defaultScheduler == null) {	1✔
277	defaultScheduler =	1✔
278	Executors.newScheduledThreadPool(	1✔
279	// Two threads seem needed here, so that reading PID decisions doesn't interfere
280	// overly with firing off scheduled tasks or one another.
281	2,
282	r -> {
283	Thread t = new Thread(r);	1✔
284	t.setName("ResourceBasedSlotSupplier.scheduler");	1✔
285	t.setDaemon(true);	1✔
286	return t;	1✔
287	});
288	}
289	return defaultScheduler;	1✔
290	}
291	}
292	}

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