org.axonframework.eventhandling

Class TrackingEventProcessor

java.lang.Object

org.axonframework.eventhandling.AbstractEventProcessor

org.axonframework.eventhandling.TrackingEventProcessor

All Implemented Interfaces:

EventProcessor, MessageHandlerInterceptorSupport<EventMessage<?>>

public class TrackingEventProcessor
extends AbstractEventProcessor

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 remove the tracking token of this processor from the TokenStore. To replay from a given point first update the entry for this processor in the TokenStore before starting this processor.

Note, the name of the EventProcessor 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

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	TrackingEventProcessor.Builder Builder class to instantiate a TrackingEventProcessor.
protected static class	TrackingEventProcessor.State Enum representing the possible states of the Processor

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	TrackingEventProcessor(TrackingEventProcessor.Builder builder) Instantiate a TrackingEventProcessor based on the fields contained in the TrackingEventProcessor.Builder.

Method Summary

Modifier and Type	Method and Description
int	activeProcessorThreads() Returns an approximation of the number of threads currently processing events.
int	availableProcessorThreads() Returns the number of threads this processor has available to assign segments.
static TrackingEventProcessor.Builder	builder() Instantiate a Builder to be able to create a TrackingEventProcessor.
protected boolean	canHandle(EventMessage<?> eventMessage, Collection<Segment> segments) Indicates whether any of the components handling events for this Processor are able to handle the given eventMessage for any of the given segments.
protected void	doSleepFor(long millisToSleep) Instructs the current Thread to sleep until the given deadline.
StreamableMessageSource<? extends TrackedEventMessage<?>>	getMessageSource() Returns the StreamableMessageSource this processor is using
protected TrackingEventProcessor.State	getState() Get the state of the event processor.
String	getTokenStoreIdentifier() Returns the unique identifier of the TokenStore used by this EventProcessor.
boolean	isError() Indicates whether the processor has been paused due to an error.
boolean	isRunning() Indicates whether this processor is currently running (i.e.
CompletableFuture<Boolean>	mergeSegment(int segmentId) Instruct the processor to merge the segment with given segmentId back with the segment that it was originally split from.
protected void	processingLoop(Segment segment) Fetch and process event batches continuously for as long as the processor is not shutting down.
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.
Map<Integer,EventTrackerStatus>	processingStatus() Returns the status for each of the Segments processed by the current processor.
void	releaseSegment(int segmentId) Instructs the processor to release the segment with given segmentId.
void	releaseSegment(int segmentId, long blacklistDuration, TimeUnit unit) Instructs the processor to release the segment with given segmentId.
void	resetTokens() Resets tokens to their initial state.
void	resetTokens(Function<StreamableMessageSource<TrackedEventMessage<?>>,TrackingToken> initialTrackingTokenSupplier) Reset tokens to the position as return by the given initialTrackingTokenSupplier.
<R> void	resetTokens(Function<StreamableMessageSource<TrackedEventMessage<?>>,TrackingToken> initialTrackingTokenSupplier, R resetContext) Reset tokens to the position as return by the given initialTrackingTokenSupplier.
<R> void	resetTokens(R resetContext) Resets tokens to their initial state.
void	resetTokens(TrackingToken startPosition) Resets tokens to the given startPosition.
<R> void	resetTokens(TrackingToken startPosition, R resetContext) Resets tokens to the given startPosition.
void	shutDown() Shuts down the processor.
CompletableFuture<Void>	shutdownAsync() Initiates a shutdown, providing a CompletableFuture that completes when the shutdown process is finished.
CompletableFuture<Boolean>	splitSegment(int segmentId) Instruct the processor to split the segment with given segmentId into two segments, allowing an additional thread to start processing events concurrently.
void	start() Start this processor.
protected void	startSegmentWorkers() Starts workers for a number of segments.
boolean	supportsReset() Indicates whether this tracking processor supports a "reset".

Methods inherited from class org.axonframework.eventhandling.AbstractEventProcessor

canHandle, canHandleType, eventHandlerInvoker, getHandlerInterceptors, getName, processInUnitOfWork, processInUnitOfWork, registerHandlerInterceptor, reportIgnored, toString

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Constructor Detail

TrackingEventProcessor

protected TrackingEventProcessor(TrackingEventProcessor.Builder builder)

Instantiate a TrackingEventProcessor based on the fields contained in the TrackingEventProcessor.Builder.

Will assert that the Event Processor name, EventHandlerInvoker, StreamableMessageSource, TokenStore and TransactionManager are not null, and will throw an AxonConfigurationException if any of them is null.

Parameters:

builder - the TrackingEventProcessor.Builder used to instantiate a TrackingEventProcessor instance

Method Detail

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 and the TrackingEventProcessorConfiguration to a TrackingEventProcessorConfiguration.forSingleThreadedProcessing() call. 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

start

@StartHandler(phase=1073741823)
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.

splitSegment

public CompletableFuture<Boolean> splitSegment(int segmentId)

Instruct the processor to split the segment with given segmentId into two segments, allowing an additional thread to start processing events concurrently.

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 thread owned by this processor instance.

Parameters:

segmentId - The identifier of the segment to split

Returns:

a CompletableFuture providing the result of the split operation

getTokenStoreIdentifier

public String getTokenStoreIdentifier()

Returns the unique identifier of the TokenStore used by this EventProcessor.

Returns:

the unique identifier of the TokenStore used by this EventProcessor

Throws:

UnableToRetrieveIdentifierException - if the tokenStore was unable to retrieve it

mergeSegment

public CompletableFuture<Boolean> mergeSegment(int segmentId)

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 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();
 

Parameters:

segmentId - The identifier of the segment to merge into this one.

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 blacklist this segment for twice the token claim interval, to ensure it is not immediately reclaimed.

Parameters:

segmentId - the id of the segment to be blacklisted

releaseSegment

public void releaseSegment(int segmentId,
                           long blacklistDuration,
                           TimeUnit unit)

Instructs the processor to release the segment with given segmentId. This will also blacklist this segment for the given blacklistDuration, to ensure it is not immediately reclaimed. Note that this will override any previous blacklist 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, it will be blacklisted nonetheless.

Parameters:

segmentId - the id of the segment to be blacklisted

blacklistDuration - the amount of time to blacklist this segment for processing by this processor instance

unit - the unit of time used to express the blacklistDuration

resetTokens

public void resetTokens()

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.

resetTokens

public <R> void resetTokens(R resetContext)

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.

Type Parameters:

R - the type of the provided resetContext

Parameters:

resetContext - a R used to support the reset operation

resetTokens

public void resetTokens(Function<StreamableMessageSource<TrackedEventMessage<?>>,TrackingToken> initialTrackingTokenSupplier)

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.

Parameters:

initialTrackingTokenSupplier - A function returning the token representing the position to reset to

resetTokens

public <R> void resetTokens(Function<StreamableMessageSource<TrackedEventMessage<?>>,TrackingToken> initialTrackingTokenSupplier,
                            R resetContext)

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.

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(TrackingToken startPosition)

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.

Parameters:

startPosition - the token representing the position to reset the processor to

resetTokens

public <R> void resetTokens(TrackingToken startPosition,
                            R resetContext)

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.

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

Indicates whether this tracking processor supports a "reset". Generally, a reset is supported if at least one of the event handlers assigned to this processor supports it, and no handlers explicitly prevent the resets.

Returns:

true if resets are supported, false otherwise

isRunning

public boolean isRunning()

Indicates whether this processor is currently running (i.e. consuming events from a stream).

Returns:

true when running, otherwise false

isError

public boolean isError()

Indicates whether the processor has been paused due to an error. In such case, the processor has forcefully paused, as it wasn't able to automatically recover.

Note that this method also returns false when the processor was stooped using shutDown().

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.

shutdownAsync

@ShutdownHandler(phase=1073741823)
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.

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.

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

Returns the status for each of the Segments processed by the current processor. The key of the map represents the SegmentID 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.

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