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 it's 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

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	PooledStreamingEventProcessor.Builder Builder class to instantiate a PooledStreamingEventProcessor.

Nested classes/interfaces inherited from interface org.axonframework.lifecycle.Lifecycle

Lifecycle.LifecycleHandler, Lifecycle.LifecycleRegistry

Field Summary

Fields inherited from class org.axonframework.eventhandling.AbstractEventProcessor

spanFactory

Constructor Summary

Constructors

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

Method Summary

Modifier and Type	Method and Description
static PooledStreamingEventProcessor.Builder	builder() Instantiate a Builder to be able to create a PooledStreamingEventProcessor.
String	getTokenStoreIdentifier() Returns the unique identifier of the TokenStore used by this StreamingEventProcessor.
boolean	isError() Indicates whether the processor has been shut down due to an error.
boolean	isRunning() Indicates whether this processor is currently running (i.e.
int	maxCapacity() Specifies the maximum amount of segments this EventProcessor can process at the same time.
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.
Map<Integer,EventTrackerStatus>	processingStatus() Returns the status for each of the segments processed by this processor as EventTrackerStatus instances.
void	registerLifecycleHandlers(Lifecycle.LifecycleRegistry handle) Registers the activities to be executed in the various phases of an application's lifecycle.
void	releaseSegment(int segmentId) Instructs the processor to release the segment with given segmentId.
void	releaseSegment(int segmentId, long releaseDuration, 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() Stops processing events.
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 process to start processing events from it.
void	start() Start processing events.
boolean	supportsReset() Indicates whether this StreamingEventProcessor 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

Methods inherited from interface org.axonframework.eventhandling.StreamingEventProcessor

isReplaying

Methods inherited from interface org.axonframework.eventhandling.EventProcessor

getHandlerInterceptors, getName

Methods inherited from interface org.axonframework.messaging.MessageHandlerInterceptorSupport

registerHandlerInterceptor

Constructor Detail

PooledStreamingEventProcessor

protected PooledStreamingEventProcessor(PooledStreamingEventProcessor.Builder builder)

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

Will assert the following for their presence prior to constructing this processor:

• The Event Processor's name.
• An EventHandlerInvoker.
• A StreamableMessageSource.
• A TokenStore.
• A TransactionManager.
• A ScheduledExecutorService for coordination.
• A ScheduledExecutorService to process work packages.

If any of these is not present or does no comply to the requirements an AxonConfigurationException is thrown.

Parameters:

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

Method Detail

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 RollbackConfigurationType defaults to a RollbackConfigurationType.ANY_THROWABLE.
• The ErrorHandler is defaulted to a PropagatingErrorHandler.
• The MessageMonitor defaults to a NoOpMessageMonitor.
• The initialSegmentCount defaults to 16.
• The initialToken function defaults to StreamableMessageSource.createTailToken().
• The tokenClaimInterval defaults to 5000 milliseconds.
• The maxCapacity (used by maxCapacity()) defaults to Short.MAX_VALUE.
• The claimExtensionThreshold defaults to 5000 milliseconds.
• The batchSize defaults to 1.
• The Clock defaults to GenericEventMessage.clock.
• The SpanFactory defaults to NoOpSpanFactory.

The following fields of this builder are hard requirements and as such should be provided:

• The name of this EventProcessor.
• An EventHandlerInvoker which will be given the events handled by this processor
• A StreamableMessageSource used to retrieve events.
• A TokenStore to store the progress of this processor in.
• A ScheduledExecutorService to coordinate events and segment operations.
• A ScheduledExecutorService to process work packages.

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:

Lifecycle.LifecycleRegistry.onShutdown(int, Runnable), LifecycleRegistry#onShutdown(int, LifecycleHandler), Lifecycle.LifecycleRegistry.onStart(int, Runnable), LifecycleRegistry#onStart(int, LifecycleHandler)

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

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