diff --git a/temporal-sdk/src/test/java/io/temporal/worker/TerminatedWorkflowStickyCacheReproducer.java b/temporal-sdk/src/test/java/io/temporal/worker/TerminatedWorkflowStickyCacheReproducer.java new file mode 100644 index 0000000000..e93bd20abc --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/worker/TerminatedWorkflowStickyCacheReproducer.java @@ -0,0 +1,92 @@ +package io.temporal.worker; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.testing.TestEnvironmentOptions; +import io.temporal.testing.TestWorkflowEnvironment; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.util.ArrayList; +import java.util.List; +import org.junit.Test; + +/** + * Reproducer: externally terminated workflows are retained by the sticky cache until LRU capacity + * evicts them. The server dispatches no workflow task on termination, so the worker never learns + * the run is closed; each dead run keeps its full cached state and its workflow thread. + * + *
Observed (sdk-java 1.34, 300 terminated runs holding a 1 MB char[] each, in-process test
+ * server): heap 18 -> 510 MB, 164 workflow threads pinned.
+ */
+public class TerminatedWorkflowStickyCacheReproducer {
+
+ @WorkflowInterface
+ public interface LeakWorkflow {
+ @WorkflowMethod
+ String run(int ballastKb);
+ }
+
+ public static class LeakWorkflowImpl implements LeakWorkflow {
+ private char[] ballast;
+
+ @Override
+ public String run(int ballastKb) {
+ ballast = new char[ballastKb * 1024];
+ Workflow.await(() -> ballast.length == 0);
+ return "done";
+ }
+ }
+
+ private static long heapMb() {
+ System.gc();
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ Runtime rt = Runtime.getRuntime();
+ return (rt.totalMemory() - rt.freeMemory()) / 1048576;
+ }
+
+ private static long workflowThreads() {
+ return Thread.getAllStackTraces().keySet().stream()
+ .filter(t -> t.getName().contains("workflow-method"))
+ .count();
+ }
+
+ @Test
+ public void terminatedWorkflowsStayInCacheUntilLruCapacity() throws Exception {
+ int count = 300;
+ int ballastKb = 1024;
+
+ TestWorkflowEnvironment env =
+ TestWorkflowEnvironment.newInstance(TestEnvironmentOptions.newBuilder().build());
+ env.newWorker("leak-lab").registerWorkflowImplementationTypes(LeakWorkflowImpl.class);
+ env.start();
+ WorkflowClient client = env.getWorkflowClient();
+
+ System.out.printf("baseline: heap %d MB, %d workflow threads%n", heapMb(), workflowThreads());
+
+ List