Package org.axonframework.eventhandling

Class TrackingEventProcessor

java.lang.Object
org.axonframework.eventhandling.AbstractEventProcessor
org.axonframework.eventhandling.TrackingEventProcessor
All Implemented Interfaces:
EventProcessor, StreamingEventProcessor, Lifecycle, MessageHandlerInterceptorSupport<EventMessage<?>>
public class TrackingEventProcessor extends AbstractEventProcessor implements StreamingEventProcessor, Lifecycle
EventProcessor implementation that tracks events from a StreamableMessageSource.

A supplied TokenStore allows the EventProcessor to keep track of its position in the event log. After processing an event batch the EventProcessor updates its tracking token in the TokenStore.

A TrackingEventProcessor is able to continue processing from the last stored token when it is restarted. It is also capable of replaying events from any starting token. To replay the entire event log, simply invoke resetTokens() on this processor to adjust the positions of the TrackingToken(s) within the TokenStore. To replay from a specific point, resetTokens(Function) can be utilized to define the new point to start at.

Note, the AbstractEventProcessor.getName() of this StreamingEventProcessor is used to obtain the tracking token from the TokenStore, so take care when renaming a TrackingEventProcessor.

Since:
3.0
Author:
Rene de Waele, Christophe Bouhier

  • Constructor Details

  • Method Details

    • builder

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

      The RollbackConfigurationType defaults to a RollbackConfigurationType.ANY_THROWABLE, the ErrorHandler is defaulted to a PropagatingErrorHandler, the MessageMonitor defaults to a NoOpMessageMonitor, the TrackingEventProcessorConfiguration to a TrackingEventProcessorConfiguration.forSingleThreadedProcessing() call, and the EventProcessorSpanFactory to a DefaultEventProcessorSpanFactory backed by a NoOpSpanFactory. The Event Processor name, EventHandlerInvoker, StreamableMessageSource, TokenStore and TransactionManager are hard requirements and as such should be provided.

      Returns:
      a Builder to be able to create a TrackingEventProcessor

    • 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()
      Start this processor. The processor will open an event stream on its message source in a new thread using StreamableMessageSource.openStream(TrackingToken). The TrackingToken used to open the stream will be fetched from the TokenStore.

      Upon start up of an application, this method will be invoked in the Phase.INBOUND_EVENT_CONNECTORS phase.

      Specified by:
      start in interface EventProcessor

    • 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

    • 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

    • 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

    • processingLoop

      protected void processingLoop(Segment segment)
      Fetch and process event batches continuously for as long as the processor is not shutting down. The processor will process events in batches. The maximum size of size of each event batch is configurable.

      Events with the same tracking token (which is possible as result of upcasting) should always be processed in the same batch. In those cases the batch size may be larger than the one configured.

      Parameters:
      segment - The Segment of the Stream that should be processed.

    • processingSegments

      protected Set<Segment> processingSegments(TrackingToken token, Segment segment)
      Indicates whether the eventMessage identified with given token should be processed as part of the given segment. This implementation is away of merge tokens and will recursively detect the (sub)segment in which an event should be handled.
      Parameters:
      token - The token to check segment validity for
      segment - The segment to process the event in
      Returns:
      true if this event should be handled, otherwise false

    • canHandle

      protected boolean canHandle(EventMessage<?> eventMessage, Collection<Segment> segments) throws Exception
      Indicates whether any of the components handling events for this Processor are able to handle the given eventMessage for any of the given segments.
      Parameters:
      eventMessage - The message to handle
      segments - The segments to handle the message in
      Returns:
      whether the given message should be handled as part of anyof the give segments
      Throws:
      Exception - when an exception occurs evaluating the message

    • releaseSegment

      public void releaseSegment(int segmentId)
      Instructs the processor to release the segment with given segmentId.

      This will also ignore the specified this segment for "re-claiming" for twice the TrackingEventProcessorConfiguration.getTokenClaimInterval() token claim interval.

      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)
      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.

      This method will put the segment on a non-claim map. During the next iteration of the processingLoop(Segment) the segments will be unclaimed and the worker stopped if it is found to be in this map. This means it can take up to the batch processing time, or up to the eventAvailabilityTimeout if there are no events in the stream for the segment to be unclaimed.

      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)
      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.

      This method will add an instruction for the TrackingEventProcessor.WorkerLauncher and set a flag that interrupts its sleep to reduce the time it takes for the processor to claim the segment. Note that a thread has to be available for a segment to start processing. The result will be false if there is none 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

    • 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

    • 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

    • 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

    • getMessageSource

      public StreamableMessageSource<? extends TrackedEventMessage<?>> getMessageSource()
      Returns the StreamableMessageSource this processor is using
      Returns:
      StreamableMessageSource

    • shutDown

      public void shutDown()
      Shuts down the processor. Blocks until shutdown is complete.
      Specified by:
      shutDown in interface EventProcessor

    • shutdownAsync

      public CompletableFuture<Void> shutdownAsync()
      Initiates a shutdown, providing a CompletableFuture that completes when the shutdown process is finished.

      Will be shutdown on the Phase.INBOUND_EVENT_CONNECTORS phase.

      Specified by:
      shutdownAsync in interface EventProcessor
      Returns:
      a CompletableFuture that completes when the shutdown process is finished.

    • availableProcessorThreads

      public int availableProcessorThreads()
      Returns the number of threads this processor has available to assign segments. These threads may or may not already be active.
      Returns:
      the number of threads this processor has available to assign segments.

    • maxCapacity

      public int maxCapacity()
      Description copied from interface: StreamingEventProcessor
      Specifies the maximum amount of segments this EventProcessor can process at the same time.
      Specified by:
      maxCapacity in interface StreamingEventProcessor
      Returns:
      the maximum amount of segments this EventProcessor can process at the same time

    • activeProcessorThreads

      public int activeProcessorThreads()
      Returns an approximation of the number of threads currently processing events.
      Returns:
      an approximation of the number of threads currently processing events

    • 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

    • getState

      protected TrackingEventProcessor.State getState()
      Get the state of the event processor. This will indicate whether or not the processor has started or is shutting down.
      Returns:
      the processor state

    • startSegmentWorkers

      protected void startSegmentWorkers()
      Starts workers for a number of segments. When only the root segment exists in the TokenStore, it will be split in multiple segments as configured by the TrackingEventProcessorConfiguration.andInitialSegmentsCount(int), otherwise the existing segments in the TokenStore will be used.

      An attempt will be made to instantiate a worker for each segment. This will succeed when the number of threads matches the requested segments. The number of active threads can be configured with TrackingEventProcessorConfiguration.forParallelProcessing(int). When insufficient threads are available to serve the number of segments, it will result in some segments not being processed.

    • doSleepFor

      protected void doSleepFor(long millisToSleep)
      Instructs the current Thread to sleep until the given deadline. This method may be overridden to check for flags that have been set to return earlier than the given deadline.

      The default implementation will sleep in blocks of 100ms, intermittently checking for the processor's state. Once the processor stops running, this method will return immediately (after detecting the state change).

      Parameters:
      millisToSleep - The number of milliseconds to sleep

    • doSleepFor

      protected void doSleepFor(long millisToSleep, AtomicBoolean interruptFlag)
      Instructs the current Thread to sleep until the given deadline. This method may be overridden to check for flags that have been set to return earlier than the given deadline.

      The default implementation will sleep in blocks of 100ms, intermittently checking for the processor's state. Once the processor stops running, this method will return immediately (after detecting the state change).

      This method will also return when the given interruptFlag is set to true. This can facilitate immediately needed actions for the sleeping thread.

      Parameters:
      millisToSleep - The number of milliseconds to sleep
      interruptFlag - A flag that is set when the thread should stop sleeping