From 785513f30cbb74ecb1cafdebe6d7980f30ffafe6 Mon Sep 17 00:00:00 2001 From: Edward Amsden Date: Thu, 8 Oct 2026 20:12:58 -0500 Subject: [PATCH] Drain Spring Boot workers during context shutdown --- CHANGELOG.md | 3 + .../io/temporal/worker/WorkerFactory.java | 7 +- temporal-spring-boot-autoconfigure/README.md | 18 +- .../build.gradle | 3 +- .../NonRootBeanPostProcessor.java | 4 + .../NonRootNamespaceAutoConfiguration.java | 11 - .../RootNamespaceAutoConfiguration.java | 10 +- .../internal/WorkerFactoryLifecycle.java | 67 ++++ .../WorkerFactoryLifecycleTest.java | 107 ++++++ .../WorkerFactorySignalShutdownTest.java | 317 ++++++++++++++++++ 10 files changed, 532 insertions(+), 15 deletions(-) create mode 100644 temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/internal/WorkerFactoryLifecycle.java create mode 100644 temporal-spring-boot-autoconfigure/src/test/java/io/temporal/spring/boot/autoconfigure/WorkerFactoryLifecycleTest.java create mode 100644 temporal-spring-boot-autoconfigure/src/test/java/io/temporal/spring/boot/autoconfigure/WorkerFactorySignalShutdownTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index af4bac6c19..9d364ae89d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,9 @@ to docs, or any other relevant information. specific to the Temporal CLI dev server implementation and may no longer be valid if that implementation changes. ### Fixed +- Spring Boot now drains root and non-root workers during application context shutdown, including + SIGINT and SIGTERM, within `spring.lifecycle.timeout-per-shutdown-phase`. Closing a child context + no longer shuts down its parent's non-root workers. - Test server now honors retry expiration deadlines that fall exactly on a whole second. Previously such deadlines were ignored and retries were scheduled past them instead of failing with `RETRY_STATE_TIMEOUT`. diff --git a/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java index e7b83034d5..e0046255af 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java @@ -45,7 +45,12 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -/** Maintains worker creation and lifecycle. */ +/** + * Maintains worker creation and lifecycle. Applications are responsible for calling {@link + * #shutdown()} and {@link #awaitTermination(long, TimeUnit)} before exiting. For example, a JVM + * shutdown hook can drain workers on SIGINT and SIGTERM. Hosting integrations such as Temporal + * Spring Boot manage this lifecycle for the application. + */ public final class WorkerFactory { private static final Logger log = LoggerFactory.getLogger(WorkerFactory.class); diff --git a/temporal-spring-boot-autoconfigure/README.md b/temporal-spring-boot-autoconfigure/README.md index a8dc71c98a..2bd284d594 100644 --- a/temporal-spring-boot-autoconfigure/README.md +++ b/temporal-spring-boot-autoconfigure/README.md @@ -2,6 +2,22 @@ For documentation on the Temporal Spring Boot Integration, please visit [https://docs.temporal.io/develop/java/spring-boot-integration](https://docs.temporal.io/develop/java/spring-boot-integration) +## Worker shutdown + +Closing the Spring application context initiates graceful shutdown for workers in the root and +non-root namespaces and waits for in-flight tasks before destroying their dependencies. With +Spring Boot's JVM shutdown hook enabled, both SIGINT and SIGTERM follow this path on POSIX systems. +Worker startup still follows `spring.temporal.start-workers` and its per-namespace overrides. + +The wait is bounded by `spring.lifecycle.timeout-per-shutdown-phase` (30 seconds by default). +For example, set it to `60s` to allow more time for activities to finish. When the timeout expires, +Spring continues closing the context; the integration does not interrupt activities with +`shutdownNow()`. Allow enough process termination time for the configured grace period. + +Applications that disable Spring Boot's JVM shutdown hook must close the application context +themselves. Standalone `WorkerFactory` applications manage their own shutdown; a JVM shutdown hook +should call `shutdown()` followed by `awaitTermination(timeout, unit)` before closing service stubs. + # Running Multiple Name Space (experimental) Along with the root namespace, you can configure multiple non-root namespaces in the application.yml file. Different namespaces can have different configurations including but not limited to different connection options, registered workflows/activities, data converters etc. @@ -60,4 +76,4 @@ annotation with the bean name corresponding to the namespace alias + `WorkflowCl @Resource(name = "unassignWorkflowClient") private WorkflowClient unassignWorkflowClient; -``` \ No newline at end of file +``` diff --git a/temporal-spring-boot-autoconfigure/build.gradle b/temporal-spring-boot-autoconfigure/build.gradle index 651d8c7c51..577d49c87e 100644 --- a/temporal-spring-boot-autoconfigure/build.gradle +++ b/temporal-spring-boot-autoconfigure/build.gradle @@ -65,7 +65,8 @@ dependencies { tasks.test { useJUnitPlatform() + systemProperty 'temporal.test.classpath', sourceSets.test.runtimeClasspath.asPath testLogging { events("passed", "skipped", "failed") } -} \ No newline at end of file +} diff --git a/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/NonRootBeanPostProcessor.java b/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/NonRootBeanPostProcessor.java index 2aeef3b18d..5ea34da33a 100644 --- a/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/NonRootBeanPostProcessor.java +++ b/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/NonRootBeanPostProcessor.java @@ -14,6 +14,7 @@ import io.temporal.serviceclient.WorkflowServiceStubsOptions; import io.temporal.serviceclient.WorkflowServiceStubsPlugin; import io.temporal.spring.boot.TemporalOptionsCustomizer; +import io.temporal.spring.boot.autoconfigure.internal.WorkerFactoryLifecycle; import io.temporal.spring.boot.autoconfigure.properties.ConnectionProperties; import io.temporal.spring.boot.autoconfigure.properties.NonRootNamespaceProperties; import io.temporal.spring.boot.autoconfigure.properties.TemporalProperties; @@ -193,6 +194,9 @@ private void injectBeanByNonRootNamespace(NonRootNamespaceProperties ns) { beanFactory.registerSingleton( beanPrefix + ScheduleClient.class.getSimpleName(), scheduleClient); beanFactory.registerSingleton(beanPrefix + WorkerFactory.class.getSimpleName(), workerFactory); + beanFactory.registerSingleton( + beanPrefix + WorkerFactoryLifecycle.class.getSimpleName(), + new WorkerFactoryLifecycle(workerFactory)); } @Override diff --git a/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/NonRootNamespaceAutoConfiguration.java b/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/NonRootNamespaceAutoConfiguration.java index 7a70e2ebaf..3963c3cf00 100644 --- a/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/NonRootNamespaceAutoConfiguration.java +++ b/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/NonRootNamespaceAutoConfiguration.java @@ -23,7 +23,6 @@ import org.springframework.context.annotation.Conditional; import org.springframework.context.annotation.Lazy; import org.springframework.context.event.ApplicationContextEvent; -import org.springframework.context.event.ContextClosedEvent; import org.springframework.context.event.ContextRefreshedEvent; @AutoConfiguration( @@ -70,8 +69,6 @@ public void onApplicationEvent(ApplicationContextEvent event) { if (event instanceof ContextRefreshedEvent) { onStart(); } - } else if (event instanceof ContextClosedEvent) { - onStop(); } } @@ -102,14 +99,6 @@ private void onStart() { }); } - private void onStop() { - this.executeByNamespace( - (nonRootNamespaceProperties, workersTemplate) -> { - log.info("shutdown workers for non-root namespace"); - workersTemplate.getWorkerFactory().shutdown(); - }); - } - private void executeByNamespace( BiConsumer consumer) { if (temporalProperties.getNamespaces() == null) { diff --git a/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/RootNamespaceAutoConfiguration.java b/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/RootNamespaceAutoConfiguration.java index 9ff3908bb5..6e4a79fc43 100644 --- a/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/RootNamespaceAutoConfiguration.java +++ b/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/RootNamespaceAutoConfiguration.java @@ -14,6 +14,7 @@ import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.serviceclient.WorkflowServiceStubsPlugin; import io.temporal.spring.boot.TemporalOptionsCustomizer; +import io.temporal.spring.boot.autoconfigure.internal.WorkerFactoryLifecycle; import io.temporal.spring.boot.autoconfigure.properties.TemporalProperties; import io.temporal.spring.boot.autoconfigure.template.ClientTemplate; import io.temporal.spring.boot.autoconfigure.template.NamespaceTemplate; @@ -42,6 +43,7 @@ import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.ApplicationListener; +import org.springframework.context.SmartLifecycle; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Conditional; import org.springframework.context.annotation.DependsOn; @@ -202,13 +204,19 @@ public WorkersTemplate workersTemplate( } @Primary - @Bean(name = "temporalWorkerFactory", destroyMethod = "shutdown") + @Bean(name = "temporalWorkerFactory", destroyMethod = "") @Conditional(WorkersPresentCondition.class) public WorkerFactory workerFactory( @Qualifier("temporalWorkersTemplate") WorkersTemplate workersTemplate) { return workersTemplate.getWorkerFactory(); } + @Bean + @Conditional(WorkersPresentCondition.class) + public SmartLifecycle temporalWorkerFactoryLifecycle(WorkerFactory workerFactory) { + return new WorkerFactoryLifecycle(workerFactory); + } + @Primary @Bean(name = "temporalWorkers") @Conditional(WorkersPresentCondition.class) diff --git a/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/internal/WorkerFactoryLifecycle.java b/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/internal/WorkerFactoryLifecycle.java new file mode 100644 index 0000000000..58532785b3 --- /dev/null +++ b/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/internal/WorkerFactoryLifecycle.java @@ -0,0 +1,67 @@ +package io.temporal.spring.boot.autoconfigure.internal; + +import io.temporal.worker.WorkerFactory; +import java.util.concurrent.TimeUnit; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.context.SmartLifecycle; + +/** Drains workers before Spring destroys their dependencies. */ +public final class WorkerFactoryLifecycle implements SmartLifecycle, DisposableBean { + private final WorkerFactory workerFactory; + + public WorkerFactoryLifecycle(WorkerFactory workerFactory) { + this.workerFactory = workerFactory; + } + + @Override + public boolean isAutoStartup() { + return false; + } + + @Override + public void start() { + // Worker startup is controlled by the existing application event listeners. + } + + @Override + public boolean isRunning() { + return !workerFactory.isTerminated(); + } + + @Override + public void stop() { + shutdown(); + workerFactory.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + } + + @Override + public void stop(Runnable callback) { + shutdown(); + Thread waiter = + new Thread( + () -> { + try { + workerFactory.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + } finally { + callback.run(); + } + }, + "temporal-worker-shutdown-" + + workerFactory.getWorkflowClient().getOptions().getNamespace()); + // Spring's lifecycle processor bounds the wait with its shutdown phase timeout. + waiter.setDaemon(true); + waiter.start(); + } + + @Override + public void destroy() { + // Bean destruction also runs when context initialization fails before lifecycle shutdown. + shutdown(); + } + + private void shutdown() { + if (!workerFactory.isShutdown()) { + workerFactory.shutdown(); + } + } +} diff --git a/temporal-spring-boot-autoconfigure/src/test/java/io/temporal/spring/boot/autoconfigure/WorkerFactoryLifecycleTest.java b/temporal-spring-boot-autoconfigure/src/test/java/io/temporal/spring/boot/autoconfigure/WorkerFactoryLifecycleTest.java new file mode 100644 index 0000000000..bfea659587 --- /dev/null +++ b/temporal-spring-boot-autoconfigure/src/test/java/io/temporal/spring/boot/autoconfigure/WorkerFactoryLifecycleTest.java @@ -0,0 +1,107 @@ +package io.temporal.spring.boot.autoconfigure; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.spring.boot.autoconfigure.internal.WorkerFactoryLifecycle; +import io.temporal.worker.WorkerFactory; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.springframework.beans.BeansException; +import org.springframework.context.support.DefaultLifecycleProcessor; +import org.springframework.context.support.GenericApplicationContext; + +@Timeout(10) +class WorkerFactoryLifecycleTest { + @Test + void boundsContextCloseWithoutInterruptingTasksOrStartingWorkers() throws Exception { + WorkerFactory factory = mock(WorkerFactory.class); + WorkflowClient client = mock(WorkflowClient.class); + when(factory.getWorkflowClient()).thenReturn(client); + when(client.getOptions()).thenReturn(WorkflowClientOptions.getDefaultInstance()); + CountDownLatch awaitingTermination = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(1); + AtomicBoolean interrupted = new AtomicBoolean(); + doAnswer( + invocation -> { + awaitingTermination.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + interrupted.set(true); + } finally { + finished.countDown(); + } + return null; + }) + .when(factory) + .awaitTermination(anyLong(), any()); + try (GenericApplicationContext context = new GenericApplicationContext()) { + DefaultLifecycleProcessor processor = new DefaultLifecycleProcessor(); + processor.setBeanFactory(context.getBeanFactory()); + processor.setTimeoutPerShutdownPhase(100); + context.getBeanFactory().registerSingleton("lifecycleProcessor", processor); + context + .getBeanFactory() + .registerSingleton("workerLifecycle", new WorkerFactoryLifecycle(factory)); + context.refresh(); + verify(factory, never()).start(); + context.close(); + assertTrue(awaitingTermination.await(1, TimeUnit.SECONDS)); + verify(factory).shutdown(); + verify(factory, never()).shutdownNow(); + assertEquals(1, finished.getCount()); + assertFalse(interrupted.get()); + } finally { + release.countDown(); + if (awaitingTermination.getCount() == 0) { + assertTrue(finished.await(1, TimeUnit.SECONDS)); + } + } + } + + @Test + void awaitsShutdownAlreadyInitiatedByApplication() { + WorkerFactory factory = mock(WorkerFactory.class); + WorkflowClient client = mock(WorkflowClient.class); + when(factory.getWorkflowClient()).thenReturn(client); + when(client.getOptions()).thenReturn(WorkflowClientOptions.getDefaultInstance()); + when(factory.isShutdown()).thenReturn(true); + try (GenericApplicationContext context = new GenericApplicationContext()) { + context + .getBeanFactory() + .registerSingleton("workerLifecycle", new WorkerFactoryLifecycle(factory)); + context.refresh(); + context.close(); + verify(factory, never()).shutdown(); + verify(factory).awaitTermination(anyLong(), any()); + } + } + + @Test + void shutsDownWorkersWhenContextInitializationFails() { + WorkerFactory factory = mock(WorkerFactory.class); + try (GenericApplicationContext context = new GenericApplicationContext()) { + context.registerBean( + "workerLifecycle", + WorkerFactoryLifecycle.class, + () -> new WorkerFactoryLifecycle(factory)); + context.registerBean( + "broken", + Object.class, + () -> { + throw new IllegalStateException("Context initialization failed."); + }, + definition -> definition.setDependsOn("workerLifecycle")); + assertThrows(BeansException.class, context::refresh); + verify(factory).shutdown(); + verify(factory, never()).shutdownNow(); + } + } +} diff --git a/temporal-spring-boot-autoconfigure/src/test/java/io/temporal/spring/boot/autoconfigure/WorkerFactorySignalShutdownTest.java b/temporal-spring-boot-autoconfigure/src/test/java/io/temporal/spring/boot/autoconfigure/WorkerFactorySignalShutdownTest.java new file mode 100644 index 0000000000..ea3ca26af9 --- /dev/null +++ b/temporal-spring-boot-autoconfigure/src/test/java/io/temporal/spring/boot/autoconfigure/WorkerFactorySignalShutdownTest.java @@ -0,0 +1,317 @@ +package io.temporal.spring.boot.autoconfigure; + +import static org.junit.jupiter.api.Assertions.*; + +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.activity.ActivityOptions; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import io.temporal.spring.boot.TemporalOptionsCustomizer; +import io.temporal.testing.TestWorkflowEnvironment; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; +import java.util.function.Supplier; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.condition.EnabledOnOs; +import org.junit.jupiter.api.condition.OS; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.support.GenericApplicationContext; + +@Timeout(90) +class WorkerFactorySignalShutdownTest { + @TempDir Path directory; + + @EnabledOnOs({OS.LINUX, OS.MAC}) + @ParameterizedTest + @CsvSource({"spring, INT", "spring, TERM", "plain, INT", "plain, TERM"}) + void drainsActivitiesOnSignal(String hosting, String signal) throws Exception { + runChild(hosting, signal); + } + + @Test + void drainsActivitiesOnContextCloseWithoutJvmHook() throws Exception { + runChild("spring", "close"); + } + + @Test + void closingChildContextLeavesParentWorkersRunning() throws Exception { + runChild("spring", "child-close"); + } + + private static boolean manualClose(String action) { + return "close".equals(action) || "child-close".equals(action); + } + + private void runChild(String hosting, String signal) throws Exception { + Path output = directory.resolve("output.log"); + Process child = + new ProcessBuilder( + Paths.get(System.getProperty("java.home"), "bin", "java").toString(), + "-cp", + System.getProperty("temporal.test.classpath"), + Child.class.getName(), + directory.toString(), + hosting, + signal) + .redirectErrorStream(true) + .redirectOutput(output.toFile()) + .start(); + try { + await(() -> Files.exists(directory.resolve("ready")), child, output); + if (manualClose(signal)) { + Files.createFile(directory.resolve("close")); + } else { + String pid = read(directory.resolve("pid")); + Process kill = new ProcessBuilder("kill", "-" + signal, pid).start(); + assertTrue(kill.waitFor(5, TimeUnit.SECONDS)); + assertEquals(0, kill.exitValue()); + } + await(() -> Files.exists(directory.resolve("root-shutdown")), child, output); + if ("spring".equals(hosting)) { + await(() -> Files.exists(directory.resolve("secondary-shutdown")), child, output); + } + assertTrue(child.isAlive(), read(output)); + assertFalse(Files.exists(directory.resolve("terminated"))); + Files.createFile(directory.resolve("release")); + assertTrue(child.waitFor(30, TimeUnit.SECONDS), read(output)); + assertTrue(Files.exists(directory.resolve("root-completed")), read(output)); + if ("spring".equals(hosting)) { + assertTrue(Files.exists(directory.resolve("secondary-completed")), read(output)); + } + assertTrue(Files.exists(directory.resolve("terminated")), read(output)); + if (manualClose(signal)) { + assertEquals(0, child.exitValue(), read(output)); + } + } finally { + child.destroyForcibly(); + assertTrue(child.waitFor(10, TimeUnit.SECONDS), "Child JVM did not exit."); + } + } + + private static void await(BooleanSupplier condition, Process child, Path output) + throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(40); + while (!condition.getAsBoolean() && child.isAlive() && System.nanoTime() < deadline) { + Thread.sleep(20); + } + assertTrue(condition.getAsBoolean(), read(output)); + } + + private static String read(Path path) throws IOException { + return new String(Files.readAllBytes(path), StandardCharsets.UTF_8).trim(); + } + + @WorkflowInterface + public interface DrainWorkflow { + @WorkflowMethod + void run(); + } + + public static class DrainWorkflowImpl implements DrainWorkflow { + @Override + public void run() { + Workflow.newActivityStub( + DrainActivity.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofMinutes(2)).build()) + .run(); + } + } + + @ActivityInterface + public interface DrainActivity { + @ActivityMethod + void run(); + } + + public static class Child { + private static Path directory; + private static TestWorkflowEnvironment environment; + + public static void main(String[] args) throws Exception { + directory = Paths.get(args[0]); + environment = TestWorkflowEnvironment.newInstance(); + if ("plain".equals(args[1])) { + WorkerFactory factory = environment.getWorkerFactory(); + Worker worker = factory.newWorker("root"); + worker.registerWorkflowImplementationTypes(DrainWorkflowImpl.class); + worker.registerActivitiesImplementations(activity("root", () -> factory)); + factory.start(); + Runtime.getRuntime() + .addShutdownHook( + new Thread( + () -> { + factory.shutdown(); + factory.awaitTermination(30, TimeUnit.SECONDS); + verifyTermination(factory); + environment.close(); + })); + startWorkflow(factory, "root"); + ready(); + new CountDownLatch(1).await(); + } else { + ConfigurableApplicationContext context = + new SpringApplicationBuilder(ChildConfiguration.class) + .web(WebApplicationType.NONE) + .registerShutdownHook(!manualClose(args[2])) + .run( + "--spring.config.name=signal-shutdown-test", + "--spring.temporal.test-server.enabled=false", + "--spring.temporal.connection.target=127.0.0.1:7233", + "--spring.lifecycle.timeout-per-shutdown-phase=20s", + "--spring.temporal.workers[0].task-queue=root", + "--spring.temporal.workers[0].workflow-classes[0]=" + + DrainWorkflowImpl.class.getName(), + "--spring.temporal.workers[0].activity-beans[0]=rootActivity", + "--spring.temporal.namespaces[0].namespace=default", + "--spring.temporal.namespaces[0].alias=secondary", + "--spring.temporal.namespaces[0].workers[0].task-queue=secondary", + "--spring.temporal.namespaces[0].workers[0].workflow-classes[0]=" + + DrainWorkflowImpl.class.getName(), + "--spring.temporal.namespaces[0].workers[0].activity-beans[0]=secondaryActivity"); + WorkerFactory root = context.getBean("temporalWorkerFactory", WorkerFactory.class); + WorkerFactory secondary = context.getBean("secondaryWorkerFactory", WorkerFactory.class); + assertNotSame(root, secondary); + if ("child-close".equals(args[2])) { + try (GenericApplicationContext unrelated = new GenericApplicationContext(context)) { + unrelated.refresh(); + } + assertFalse(root.isShutdown()); + assertFalse(secondary.isShutdown()); + } + startWorkflow(root, "root"); + startWorkflow(secondary, "secondary"); + ready(); + if (manualClose(args[2])) { + while (!Files.exists(directory.resolve("close"))) { + Thread.sleep(20); + } + context.close(); + } else { + new CountDownLatch(1).await(); + } + } + } + + private static DrainActivity activity(String name, Supplier factory) { + return () -> { + try { + Files.createFile(directory.resolve(name + "-started")); + while (!factory.get().isShutdown()) { + Thread.sleep(20); + } + Files.createFile(directory.resolve(name + "-shutdown")); + while (!Files.exists(directory.resolve("release"))) { + Thread.sleep(20); + } + Files.createFile(directory.resolve(name + "-completed")); + } catch (IOException | InterruptedException e) { + throw new RuntimeException(e); + } + }; + } + + private static void startWorkflow(WorkerFactory factory, String queue) { + DrainWorkflow workflow = + factory + .getWorkflowClient() + .newWorkflowStub( + DrainWorkflow.class, WorkflowOptions.newBuilder().setTaskQueue(queue).build()); + WorkflowClient.start(workflow::run); + } + + private static void ready() throws Exception { + while (!Files.exists(directory.resolve("root-started")) + || (Files.exists(directory.resolve("spring")) + && !Files.exists(directory.resolve("secondary-started")))) { + Thread.sleep(20); + } + Files.write( + directory.resolve("pid"), + ManagementFactory.getRuntimeMXBean() + .getName() + .split("@")[0] + .getBytes(StandardCharsets.UTF_8)); + Files.createFile(directory.resolve("ready")); + } + + private static void verifyTermination(WorkerFactory... factories) { + for (WorkerFactory factory : factories) { + assertTrue(factory.isShutdown()); + assertTrue(factory.isTerminated()); + } + try { + Files.createFile(directory.resolve("terminated")); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + + @EnableAutoConfiguration + public static class ChildConfiguration { + @Bean + TestWorkflowEnvironment signalTestEnvironment() throws IOException { + Files.createFile(Child.directory.resolve("spring")); + return Child.environment; + } + + @Bean + TemporalOptionsCustomizer workflowServiceStubsCustomizer( + TestWorkflowEnvironment environment) { + return builder -> + builder.setTarget(null).setChannel(environment.getWorkflowServiceStubs().getRawChannel()); + } + + @Bean + TemporalOptionsCustomizer + secondaryWorkflowServiceStubsCustomizer(TestWorkflowEnvironment environment) { + return builder -> + builder.setTarget(null).setChannel(environment.getWorkflowServiceStubs().getRawChannel()); + } + + @Bean + DrainActivity rootActivity( + @Qualifier("temporalWorkerFactory") ObjectProvider factory) { + return Child.activity("root", factory::getObject); + } + + @Bean + DrainActivity secondaryActivity( + @Qualifier("secondaryWorkerFactory") ObjectProvider factory) { + return Child.activity("secondary", factory::getObject); + } + + @Bean + DisposableBean drainVerifier( + @Qualifier("temporalWorkerFactory") WorkerFactory root, + @Qualifier("secondaryWorkerFactory") WorkerFactory secondary) { + return () -> Child.verifyTermination(root, secondary); + } + } +}