Package org.axonframework.messaging.queryhandling

Class QueryShutdownManager

java.lang.Object
org.axonframework.messaging.queryhandling.QueryShutdownManager
public class QueryShutdownManager extends Object
Manages the graceful shutdown of locally-initiated query streams, ensuring they are terminated during application shutdown.

When an application shuts down, locally-initiated subscription queries and streaming queries (those started on this node via QueryBus.subscriptionQuery(org.axonframework.messaging.queryhandling.QueryMessage, org.axonframework.messaging.core.unitofwork.ProcessingContext, int), QueryGateway.subscriptionQuery(java.lang.Object, java.lang.Class<R>, org.axonframework.messaging.core.unitofwork.ProcessingContext), or QueryGateway.streamingQuery(java.lang.Object, java.lang.Class<R>, org.axonframework.messaging.core.unitofwork.ProcessingContext)) remain open indefinitely unless explicitly closed. This prevents a web server from completing a graceful shutdown, as active connections (e.g., SSE streams) are kept alive by the open query streams.

This manager solves the problem through explicit, call-level opt-in: callers wrap their query streams or publishers using track(MessageStream) or track(Publisher), and the manager closes them when shutdown() is called.

Two shutdown policies are available:

  • Close immediately: all active subscriptions are terminated as soon as shutdown() is called.
  • Wait with grace period: the manager waits up to a configurable duration for active subscriptions to complete naturally before force-closing any that remain. This is useful when some queries are short-lived and expected to complete on their own shortly after shutdown begins.

For gateway-level tracking, use MessagingConfigurer.registerQueryGateway(String, java.util.function.Consumer), which wires the manager into Axon's lifecycle automatically: 


 MessagingConfigurer.create()
     .registerQueryGateway("sse", g -> g
         .cancellingSubscriptionQueryOnShutdown(Duration.ofSeconds(10)));
For standalone use with call-site tracking, create the manager directly and register it with Axon's lifecycle using registerShutdown(LifecycleRegistry): 

 QueryShutdownManager manager = QueryShutdownManager.withGracePeriod(Duration.ofSeconds(5));
 configurer.lifecycleRegistry(manager::registerShutdown);

When call-site tracking is needed, for example when multiple managers are defined or when using the query bus directly, wrap results explicitly with track(MessageStream) or track(Publisher): 


 // Via query gateway (returns Publisher<T>):
 Publisher<MyDto> stream = shutdownManager.track(
     queryGateway.subscriptionQuery(new MyQuery(), MyDto.class)
 );

 // Via query bus directly (returns MessageStream<T>):
 MessageStream<QueryResponseMessage> stream = shutdownManager.track(
     queryBus.subscriptionQuery(query, null, 100)
 );
Since:
5.4.0
Author:
Allard Buijze

  • Method Details

    • closeImmediately

      public static QueryShutdownManager closeImmediately()
      Creates a QueryShutdownManager that closes all active query streams immediately when shutdown() is called.
      Returns:
      a new manager configured to close immediately on shutdown

    • withGracePeriod

      public static QueryShutdownManager withGracePeriod(Duration gracePeriod)
      Creates a QueryShutdownManager that waits up to the given gracePeriod for active query streams to complete naturally before force-closing any remaining ones.
      Parameters:
      gracePeriod - the maximum time to wait for natural completion before force-closing
      Returns:
      a new manager configured to wait before closing on shutdown

    • registerShutdown

      public QueryShutdownManager registerShutdown(LifecycleRegistry registry)
      Registers shutdown() as a shutdown handler on the given registry at Phase.OUTBOUND_QUERY_CONNECTORS, the phase at which outbound query connectors shut down. At this point, no new queries can be sent, so cancelling existing subscription query streams is the natural complement.

      Returns this for fluent chaining, for example: 

      
 QueryShutdownManager manager = QueryShutdownManager.withGracePeriod(Duration.ofSeconds(5));
 configurer.lifecycleRegistry(manager::registerShutdown);
      Parameters:
      registry - the lifecycle registry to register the shutdown handler with
      Returns:
      this manager, for fluent chaining

    • track

      public <T extends Message> MessageStream<T> track(MessageStream<T> stream)
      Tracks the given stream, registering it for termination when shutdown() is called.

      The stream is automatically deregistered when it closes naturally. The caller must use the returned stream instead of the original, as the deregistration hook is attached to the returned wrapper.

      Type Parameters:
      T - the message type
      Parameters:
      stream - the query stream to track
      Returns:
      a wrapped stream that participates in this manager's lifecycle

    • track

      public <T> org.reactivestreams.Publisher<T> track(org.reactivestreams.Publisher<T> source)
      Tracks the given source publisher, registering it for termination when shutdown() is called.

      The publisher is automatically deregistered when it terminates naturally. The caller must subscribe to and use the returned Publisher instead of the original source. When shutdown is triggered, a ShutdownInProgressException is emitted as an error to any active subscriber.

      Type Parameters:
      T - the element type
      Parameters:
      source - the query publisher to track
      Returns:
      a Publisher that participates in this manager's lifecycle

    • shutdown

      public CompletableFuture<Void> shutdown()
      Shuts down all tracked query streams according to this manager's configured policy.

      If no grace period is configured, all active subscriptions are closed immediately and the returned future completes right away. Otherwise, the returned future completes once all active subscriptions have either completed naturally or been force-closed after the grace period expires.

      Returns:
      a CompletableFuture that completes when all tracked subscriptions are closed