Package org.axonframework.eventhandling.pooled

Class PooledStreamingEventProcessor

java.lang.Object
org.axonframework.eventhandling.AbstractEventProcessor
org.axonframework.eventhandling.pooled.PooledStreamingEventProcessor
All Implemented Interfaces:
EventProcessor, StreamingEventProcessor, Lifecycle, MessageHandlerInterceptorSupport<EventMessage<?>>
public class PooledStreamingEventProcessor extends AbstractEventProcessor implements StreamingEventProcessor, Lifecycle
A StreamingEventProcessor implementation which pools its resources to enhance processing speed. It utilizes a Coordinator as the means to stream events from a StreamableMessageSource and creates so-called work packages. Every work package is in charge of a Segment of the entire event stream. It is the Coordinator's job to retrieve the events from the source and provide the events to all the work packages it is in charge of.

This approach utilizes two threads pools. One to retrieve the events to provide them to the work packages and another to actual handle the events. Respectively, the coordinator thread pool and the work package thread pool. It is this approach which allows for greater parallelization and processing speed than the TrackingEventProcessor.

If no TrackingTokens are present for this processor, the PooledStreamingEventProcessor will initialize them in a given segment count. By default, it will create 16 segments, which can be configured through the PooledStreamingEventProcessor.Builder.initialSegmentCount(int).

Since:
4.5
Author:
Allard Buijze, Steven van Beelen

  • Constructor Details

  • Method Details

    • builder

      public static PooledStreamingEventProcessor.Builder builder()
      Instantiate a Builder to be able to create a PooledStreamingEventProcessor.

      Upon initialization of this builder, the following fields are defaulted:

      The following fields of this builder are hard requirements and as such should be provided:
      Returns:
      a Builder to be able to create a PooledStreamingEventProcessor

    • registerLifecycleHandlers

      public void registerLifecycleHandlers(@Nonnull Lifecycle.LifecycleRegistry handle)
      Description copied from interface: Lifecycle
      Registers the activities to be executed in the various phases of an application's lifecycle. This could either be at startup, shutdown, or both.
      Specified by:
      registerLifecycleHandlers in interface Lifecycle
      Parameters:
      handle - the lifecycle instance to register the handlers with
      See Also:

    • start

      public void start()
      Description copied from interface: EventProcessor
      Start processing events.
      Specified by:
      start in interface EventProcessor

    • shutDown

      public void shutDown()
      Description copied from interface: EventProcessor
      Stops processing events. Blocks until the shutdown is complete.
      Specified by:
      shutDown in interface EventProcessor

    • shutdownAsync

      public CompletableFuture<Void> shutdownAsync()
      Description copied from interface: EventProcessor
      Initiates a shutdown, providing a CompletableFuture that completes when the shutdown process is finished.
      Specified by:
      shutdownAsync in interface EventProcessor
      Returns:
      a CompletableFuture that completes when the shutdown process is finished.

    • isRunning

      public boolean isRunning()
      Description copied from interface: EventProcessor
      Indicates whether this processor is currently running (i.e. consuming events from its message source).
      Specified by:
      isRunning in interface EventProcessor
      Returns:
      true when running, otherwise false

    • isError

      public boolean isError()
      Description copied from interface: EventProcessor
      Indicates whether the processor has been shut down due to an error. In such case, the processor has forcefully shut down, as it wasn't able to automatically recover.

      Note that this method returns false when the processor was stopped using EventProcessor.shutDown().

      Specified by:
      isError in interface EventProcessor
      Returns:
      true when paused due to an error, otherwise false

    • getTokenStoreIdentifier

      public String getTokenStoreIdentifier()
      Description copied from interface: StreamingEventProcessor
      Returns the unique identifier of the TokenStore used by this StreamingEventProcessor.
      Specified by:
      getTokenStoreIdentifier in interface StreamingEventProcessor
      Returns:
      the unique identifier of the TokenStore used by this StreamingEventProcessor

    • releaseSegment

      public void releaseSegment(int segmentId)
      Description copied from interface: StreamingEventProcessor
      Instructs the processor to release the segment with given segmentId.
      Specified by:
      releaseSegment in interface StreamingEventProcessor
      Parameters:
      segmentId - the id of the segment to release

    • releaseSegment

      public void releaseSegment(int segmentId, long releaseDuration, TimeUnit unit)
      Description copied from interface: StreamingEventProcessor
      Instructs the processor to release the segment with given segmentId. This processor will not try to claim the given segment for the specified releaseDuration in the given unit, to ensure it is not immediately reclaimed. Note that this will override any previous release duration that existed for this segment. Providing a negative value will allow the segment to be immediately claimed.

      If the processor is not actively processing the segment with given segmentId, claiming it will be ignored for the given timeframe nonetheless.

      Specified by:
      releaseSegment in interface StreamingEventProcessor
      Parameters:
      segmentId - the id of the segment to be released for the specified releaseDuration
      releaseDuration - the amount of time to disregard segmentId for processing
      unit - the unit of time used to express the releaseDuration

    • claimSegment

      public CompletableFuture<Boolean> claimSegment(int segmentId)
      Description copied from interface: StreamingEventProcessor
      Instructs the processor to claim the segment with given segmentId and start processing it as soon as possible.

      The given segmentId must not be currently processed by a different processor instance, as that will have an active claim on the token. Claiming a segment that is already being processed will have no effect and return true.

      A true return value indicates that the segment has been claimed and will be processed by this processor. The StreamingEventProcessor may postpone start of work until after completion of this task, as long as the token has been claimed so work can be started. A return value of false indicates that the segment has not been claimed due to the token for that segment not being available.

      Specified by:
      claimSegment in interface StreamingEventProcessor
      Parameters:
      segmentId - the identifier of the segment to claim and start processing
      Returns:
      a CompletableFuture providing the result of the claim operation

    • splitSegment

      public CompletableFuture<Boolean> splitSegment(int segmentId)
      Description copied from interface: StreamingEventProcessor
      Instruct the processor to split the segment with given segmentId into two segments, allowing an additional process to start processing events from it.

      To be able to split segments, the TokenStore configured with this processor must use explicitly initialized tokens. See TokenStore.requiresExplicitSegmentInitialization(). Also, the given segmentId must be currently processed by a process owned by this processor instance.

      Specified by:
      splitSegment in interface StreamingEventProcessor
      Parameters:
      segmentId - the identifier of the segment to split
      Returns:
      a CompletableFuture providing the result of the split operation

    • mergeSegment

      public CompletableFuture<Boolean> mergeSegment(int segmentId)
      Description copied from interface: StreamingEventProcessor
      Instruct the processor to merge the segment with given segmentId back with the segment that it was originally split from. The processor must be able to claim the other segment, in order to merge it. Therefore, this other segment must not have any active claims in the TokenStore.

      The processor must currently be actively processing the segment with given segmentId.

      Use StreamingEventProcessor.releaseSegment(int) to force this processor to release any claims with tokens required to merge the segments.

      To find out which segment a given segmentId should be merged with, use the following procedure: 

           EventTrackerStatus status = processor.processingStatus().get(segmentId);
     if (status == null) {
         // this processor is not processing segmentId, and will not be able to merge
     }
     return status.getSegment().mergeableSegmentId();
      Specified by:
      mergeSegment in interface StreamingEventProcessor
      Parameters:
      segmentId - the identifier of the segment to merge
      Returns:
      a CompletableFuture indicating whether the merge was executed successfully

    • supportsReset

      public boolean supportsReset()
      Description copied from interface: StreamingEventProcessor
      Indicates whether this StreamingEventProcessor supports a "reset". Generally, a reset is supported if at least one of the Event Handling Components assigned to this processor supports it, and no handlers explicitly prevent the resets.

      This method should be invoked prior to invoking any of the StreamingEventProcessor.resetTokens() operations as an early validation.

      Specified by:
      supportsReset in interface StreamingEventProcessor
      Returns:
      true if resets are supported, false otherwise

    • resetTokens

      public void resetTokens()
      Description copied from interface: StreamingEventProcessor
      Resets tokens to their initial state. This effectively causes a replay.

      Before attempting to reset the tokens, the caller must stop this processor, as well as any instances of the same logical processor that may be running in the cluster. Failure to do so will cause the reset to fail, as a processor can only reset the tokens if it is able to claim them all.

      Specified by:
      resetTokens in interface StreamingEventProcessor

    • resetTokens

      public <R> void resetTokens(R resetContext)
      Description copied from interface: StreamingEventProcessor
      Resets tokens to their initial state. This effectively causes a replay. The given resetContext will be used to support the (optional) reset operation in an Event Handling Component.

      Before attempting to reset the tokens, the caller must stop this processor, as well as any instances of the same logical processor that may be running in the cluster. Failure to do so will cause the reset to fail, as a processor can only reset the tokens if it is able to claim them all.

      Specified by:
      resetTokens in interface StreamingEventProcessor
      Type Parameters:
      R - the type of the provided resetContext
      Parameters:
      resetContext - a R used to support the reset operation

    • resetTokens

      public void resetTokens(@Nonnull Function<StreamableMessageSource<TrackedEventMessage<?>>,TrackingToken> initialTrackingTokenSupplier)
      Description copied from interface: StreamingEventProcessor
      Reset tokens to the position as return by the given initialTrackingTokenSupplier. This effectively causes a replay since that position.

      Note that the new token must represent a position that is before the current position of the processor.

      Before attempting to reset the tokens, the caller must stop this processor, as well as any instances of the same logical processor that may be running in the cluster. Failure to do so will cause the reset to fail, as a processor can only reset the tokens if it is able to claim them all.

      Specified by:
      resetTokens in interface StreamingEventProcessor
      Parameters:
      initialTrackingTokenSupplier - a function returning the token representing the position to reset to

    • resetTokens

      public <R> void resetTokens(@Nonnull Function<StreamableMessageSource<TrackedEventMessage<?>>,TrackingToken> initialTrackingTokenSupplier, R resetContext)
      Description copied from interface: StreamingEventProcessor
      Reset tokens to the position as return by the given initialTrackingTokenSupplier. This effectively causes a replay since that position. The given resetContext will be used to support the (optional) reset operation in an Event Handling Component.

      Note that the new token must represent a position that is before the current position of the processor.

      Before attempting to reset the tokens, the caller must stop this processor, as well as any instances of the same logical processor that may be running in the cluster. Failure to do so will cause the reset to fail, as a processor can only reset the tokens if it is able to claim them all.

      Specified by:
      resetTokens in interface StreamingEventProcessor
      Type Parameters:
      R - the type of the provided resetContext
      Parameters:
      initialTrackingTokenSupplier - a function returning the token representing the position to reset to
      resetContext - a R used to support the reset operation

    • resetTokens

      public void resetTokens(@Nonnull TrackingToken startPosition)
      Description copied from interface: StreamingEventProcessor
      Resets tokens to the given startPosition. This effectively causes a replay of events since that position.

      Note that the new token must represent a position that is before the current position of the processor.

      Before attempting to reset the tokens, the caller must stop this processor, as well as any instances of the same logical processor that may be running in the cluster. Failure to do so will cause the reset to fail, as a processor can only reset the tokens if it is able to claim them all.

      Specified by:
      resetTokens in interface StreamingEventProcessor
      Parameters:
      startPosition - the token representing the position to reset the processor to

    • resetTokens

      public <R> void resetTokens(@Nonnull TrackingToken startPosition, R resetContext)
      Description copied from interface: StreamingEventProcessor
      Resets tokens to the given startPosition. This effectively causes a replay of events since that position. The given resetContext will be used to support the (optional) reset operation in an Event Handling Component.

      Note that the new token must represent a position that is before the current position of the processor.

      Before attempting to reset the tokens, the caller must stop this processor, as well as any instances of the same logical processor that may be running in the cluster. Failure to do so will cause the reset to fail, as a processor can only reset the tokens if it is able to claim them all.

      Specified by:
      resetTokens in interface StreamingEventProcessor
      Type Parameters:
      R - the type of the provided resetContext
      Parameters:
      startPosition - the token representing the position to reset the processor to
      resetContext - a R used to support the reset operation

    • maxCapacity

      public int maxCapacity()
      Specifies the maximum amount of segments this EventProcessor can process at the same time.

      The maximum capacity of the PooledStreamingEventProcessor defaults to 32767. If required, this value can be adjusted through the PooledStreamingEventProcessor.Builder.maxClaimedSegments(int) method.

      Specified by:
      maxCapacity in interface StreamingEventProcessor
      Returns:
      the maximum amount of segments this EventProcessor can process at the same time

    • processingStatus

      public Map<Integer,EventTrackerStatus> processingStatus()
      Description copied from interface: StreamingEventProcessor
      Returns the status for each of the segments processed by this processor as EventTrackerStatus instances. The key of the Map represent the segment ids processed by this instance. The values of the returned Map represent the last known status of that segment.

      Note that the returned Map is unmodifiable, but does reflect any changes made to the status as the processor is processing Events.

      Specified by:
      processingStatus in interface StreamingEventProcessor
      Returns:
      the status for each of the segments processed by the current processor