Package org.axonframework.eventhandling.pooled

Class PooledStreamingEventProcessor.Builder

java.lang.Object

org.axonframework.eventhandling.AbstractEventProcessor.Builder

org.axonframework.eventhandling.pooled.PooledStreamingEventProcessor.Builder

Enclosing class:

PooledStreamingEventProcessor

public static class PooledStreamingEventProcessor.Builder
extends AbstractEventProcessor.Builder

Builder class to instantiate 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 a ReplayToken that starts streaming from the tail with the replay flag enabled until the head at the moment of initialization is reached.
• The tokenClaimInterval defaults to 5000 milliseconds.
• The MaxSegmentProvider (used by PooledStreamingEventProcessor.maxCapacity()) defaults to MaxSegmentProvider.maxShort().
• The claimExtensionThreshold defaults to 5000 milliseconds.
• The batchSize defaults to 1.
• The Clock defaults to GenericEventMessage.clock.
• The EventProcessorSpanFactory defaults to a DefaultEventProcessorSpanFactory backed by a NoOpSpanFactory.
• The coordinatorExtendsClaims defaults to a false.

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.

Field Summary

Fields inherited from class org.axonframework.eventhandling.AbstractEventProcessor.Builder

name

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	Builder()

Method Summary

Modifier and Type	Method	Description
PooledStreamingEventProcessor.Builder	batchSize(int batchSize)	Specifies the number of events to be processed inside a single transaction.
PooledStreamingEventProcessor	build()	Initializes a PooledStreamingEventProcessor as specified through this Builder.
PooledStreamingEventProcessor.Builder	claimExtensionThreshold(long claimExtensionThreshold)	Specifies a time in milliseconds the work packages of this processor should extend the claim on a TrackingToken.
PooledStreamingEventProcessor.Builder	clock(Clock clock)	Defines the Clock used for time dependent operation by this EventProcessor.
PooledStreamingEventProcessor.Builder	coordinatorExecutor(ScheduledExecutorService coordinatorExecutor)	Specifies the ScheduledExecutorService used by the coordinator of this PooledStreamingEventProcessor.
PooledStreamingEventProcessor.Builder	coordinatorExecutor(Function<String,ScheduledExecutorService> coordinatorExecutorBuilder)	Specifies a builder to construct a ScheduledExecutorService used by the coordinator of this PooledStreamingEventProcessor.
PooledStreamingEventProcessor.Builder	enableCoordinatorClaimExtension()	Enables the Coordinator to extend the claims of its WorkPackages.
PooledStreamingEventProcessor.Builder	errorHandler(ErrorHandler errorHandler)	Sets the ErrorHandler invoked when an UnitOfWork is rolled back during processing.
PooledStreamingEventProcessor.Builder	eventHandlerInvoker(EventHandlerInvoker eventHandlerInvoker)	Sets the EventHandlerInvoker which will handle all the individual EventMessages.
PooledStreamingEventProcessor.Builder	initialSegmentCount(int initialSegmentCount)	Sets the initial segment count used to create segments on start up.
PooledStreamingEventProcessor.Builder	initialToken(Function<StreamableMessageSource<TrackedEventMessage<?>>,TrackingToken> initialToken)	Specifies the Function used to generate the initial TrackingTokens.
PooledStreamingEventProcessor.Builder	maxClaimedSegments(int maxClaimedSegments)	Sets the maximum number of segments this instance may claim.
PooledStreamingEventProcessor.Builder	maxSegmentProvider(MaxSegmentProvider maxSegmentProvider)	Defines the maximum number of segment this StreamingEventProcessor may claim per instance.
PooledStreamingEventProcessor.Builder	messageMonitor(MessageMonitor<? super EventMessage<?>> messageMonitor)	Sets the MessageMonitor to monitor EventMessages before and after they're processed.
PooledStreamingEventProcessor.Builder	messageSource(StreamableMessageSource<TrackedEventMessage<?>> messageSource)	Sets the StreamableMessageSource (e.g. the EventStore) which this EventProcessor will track.
String	name()	Returns the name of this PooledStreamingEventProcessor.
PooledStreamingEventProcessor.Builder	name(String name)	Sets the name of this EventProcessor implementation.
PooledStreamingEventProcessor.Builder	rollbackConfiguration(RollbackConfiguration rollbackConfiguration)	Sets the RollbackConfiguration specifying the rollback behavior of the UnitOfWork while processing a batch of events.
PooledStreamingEventProcessor.Builder	spanFactory(EventProcessorSpanFactory spanFactory)	Sets the EventProcessorSpanFactory implementation to use for providing tracing capabilities.
PooledStreamingEventProcessor.Builder	spanFactory(SpanFactory spanFactory)	Deprecated.
PooledStreamingEventProcessor.Builder	tokenClaimInterval(long tokenClaimInterval)	Specifies the time in milliseconds the processor's coordinator should wait after a failed attempt to claim any segments for processing.
PooledStreamingEventProcessor.Builder	tokenStore(TokenStore tokenStore)	Sets the TokenStore used to store and fetch event tokens that enable this EventProcessor to track its progress.
PooledStreamingEventProcessor.Builder	transactionManager(TransactionManager transactionManager)	Sets the TransactionManager used when processing EventMessages.
protected void	validate()	Validates whether the fields contained in this Builder are set accordingly.
PooledStreamingEventProcessor.Builder	workerExecutor(ScheduledExecutorService workerExecutor)	Specifies the ScheduledExecutorService to be provided to the WorkPackages created by this PooledStreamingEventProcessor.
PooledStreamingEventProcessor.Builder	workerExecutor(Function<String,ScheduledExecutorService> workerExecutorBuilder)	Specifies a builder to construct a ScheduledExecutorService to be provided to the WorkPackages created by this PooledStreamingEventProcessor.
PooledStreamingEventProcessor.Builder	workerExecutorService(ScheduledExecutorService workerExecutor)	Deprecated. in favor of workerExecutor(ScheduledExecutorService)

Methods inherited from class java.lang.Object

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

Constructor Details

Builder

protected Builder()

Method Details

name

public PooledStreamingEventProcessor.Builder name(@Nonnull
 String name)

Description copied from class: AbstractEventProcessor.Builder

Sets the name of this EventProcessor implementation.

Overrides:

name in class AbstractEventProcessor.Builder

Parameters:

name - a String defining this EventProcessor implementation

Returns:

the current Builder instance, for fluent interfacing

eventHandlerInvoker

public PooledStreamingEventProcessor.Builder eventHandlerInvoker(@Nonnull
 EventHandlerInvoker eventHandlerInvoker)

Description copied from class: AbstractEventProcessor.Builder

Sets the EventHandlerInvoker which will handle all the individual EventMessages.

Overrides:

eventHandlerInvoker in class AbstractEventProcessor.Builder

Parameters:

eventHandlerInvoker - the EventHandlerInvoker which will handle all the individual EventMessages

Returns:

the current Builder instance, for fluent interfacing

rollbackConfiguration

public PooledStreamingEventProcessor.Builder rollbackConfiguration(@Nonnull
 RollbackConfiguration rollbackConfiguration)

Description copied from class: AbstractEventProcessor.Builder

Sets the RollbackConfiguration specifying the rollback behavior of the UnitOfWork while processing a batch of events.

Overrides:

rollbackConfiguration in class AbstractEventProcessor.Builder

Parameters:

rollbackConfiguration - the RollbackConfiguration specifying the rollback behavior of the UnitOfWork while processing a batch of events.

Returns:

the current Builder instance, for fluent interfacing

errorHandler

public PooledStreamingEventProcessor.Builder errorHandler(@Nonnull
 ErrorHandler errorHandler)

Description copied from class: AbstractEventProcessor.Builder

Sets the ErrorHandler invoked when an UnitOfWork is rolled back during processing. Defaults to a PropagatingErrorHandler.

Overrides:

errorHandler in class AbstractEventProcessor.Builder

Parameters:

errorHandler - the ErrorHandler invoked when an UnitOfWork is rolled back during processing

Returns:

the current Builder instance, for fluent interfacing

messageMonitor

public PooledStreamingEventProcessor.Builder messageMonitor(@Nonnull
 MessageMonitor<? super EventMessage<?>> messageMonitor)

Description copied from class: AbstractEventProcessor.Builder

Sets the MessageMonitor to monitor EventMessages before and after they're processed. Defaults to a NoOpMessageMonitor.

Overrides:

messageMonitor in class AbstractEventProcessor.Builder

Parameters:

messageMonitor - a MessageMonitor to monitor EventMessages before and after they're processed

Returns:

the current Builder instance, for fluent interfacing

spanFactory

public PooledStreamingEventProcessor.Builder spanFactory(@Nonnull
 EventProcessorSpanFactory spanFactory)

Description copied from class: AbstractEventProcessor.Builder

Sets the EventProcessorSpanFactory implementation to use for providing tracing capabilities. Defaults to a DefaultEventProcessorSpanFactory backed by a NoOpSpanFactory by default, which provides no tracing capabilities.

Overrides:

spanFactory in class AbstractEventProcessor.Builder

Parameters:

spanFactory - The SpanFactory implementation

Returns:

The current Builder instance, for fluent interfacing.

spanFactory

@Deprecated
public PooledStreamingEventProcessor.Builder spanFactory(@Nonnull
 SpanFactory spanFactory)

Deprecated.

Description copied from class: AbstractEventProcessor.Builder

Sets the SpanFactory implementation to use for providing tracing capabilities. Defaults to a NoOpSpanFactory by default, which provides no tracing capabilities.

Overrides:

spanFactory in class AbstractEventProcessor.Builder

Parameters:

spanFactory - The SpanFactory implementation

Returns:

The current Builder instance, for fluent interfacing.

messageSource

public PooledStreamingEventProcessor.Builder messageSource(@Nonnull
 StreamableMessageSource<TrackedEventMessage<?>> messageSource)

Sets the StreamableMessageSource (e.g. the EventStore) which this EventProcessor will track.

Parameters:

messageSource - the StreamableMessageSource (e.g. the EventStore) which this EventProcessor will track

Returns:

the current Builder instance, for fluent interfacing

tokenStore

public PooledStreamingEventProcessor.Builder tokenStore(@Nonnull
 TokenStore tokenStore)

Sets the TokenStore used to store and fetch event tokens that enable this EventProcessor to track its progress.

Parameters:

tokenStore - the TokenStore used to store and fetch event tokens that enable this EventProcessor to track its progress

Returns:

the current Builder instance, for fluent interfacing

transactionManager

public PooledStreamingEventProcessor.Builder transactionManager(@Nonnull
 TransactionManager transactionManager)

Sets the TransactionManager used when processing EventMessages.

Parameters:

transactionManager - the TransactionManager used when processing EventMessages

Returns:

the current Builder instance, for fluent interfacing

coordinatorExecutor

public PooledStreamingEventProcessor.Builder coordinatorExecutor(@Nonnull
 ScheduledExecutorService coordinatorExecutor)

Specifies the ScheduledExecutorService used by the coordinator of this PooledStreamingEventProcessor.

Parameters:

coordinatorExecutor - the ScheduledExecutorService to be used by the the coordinator of this PooledStreamingEventProcessor

Returns:

the current Builder instance, for fluent interfacing

coordinatorExecutor

public PooledStreamingEventProcessor.Builder coordinatorExecutor(@Nonnull
 Function<String,ScheduledExecutorService> coordinatorExecutorBuilder)

Specifies a builder to construct a ScheduledExecutorService used by the coordinator of this PooledStreamingEventProcessor.

Parameters:

coordinatorExecutorBuilder - a builder function to construct a ScheduledExecutorService, providing the PooledStreamingEventProcessor

Returns:

the current Builder instance, for fluent interfacing

workerExecutorService

@Deprecated
public PooledStreamingEventProcessor.Builder workerExecutorService(@Nonnull
 ScheduledExecutorService workerExecutor)

Deprecated.

in favor of workerExecutor(ScheduledExecutorService)

Specifies the ScheduledExecutorService to be provided to the WorkPackages created by this PooledStreamingEventProcessor.

Parameters:

workerExecutor - the ScheduledExecutorService to be provided to the WorkPackages created by this PooledStreamingEventProcessor

Returns:

the current Builder instance, for fluent interfacing

workerExecutor

public PooledStreamingEventProcessor.Builder workerExecutor(@Nonnull
 ScheduledExecutorService workerExecutor)

Specifies the ScheduledExecutorService to be provided to the WorkPackages created by this PooledStreamingEventProcessor.

Parameters:

workerExecutor - the ScheduledExecutorService to be provided to the WorkPackages created by this PooledStreamingEventProcessor

Returns:

the current Builder instance, for fluent interfacing

workerExecutor

public PooledStreamingEventProcessor.Builder workerExecutor(@Nonnull
 Function<String,ScheduledExecutorService> workerExecutorBuilder)

Specifies a builder to construct a ScheduledExecutorService to be provided to the WorkPackages created by this PooledStreamingEventProcessor.

Parameters:

workerExecutorBuilder - a builder function to construct a ScheduledExecutorService, providing the PooledStreamingEventProcessor

Returns:

the current Builder instance, for fluent interfacing

initialSegmentCount

public PooledStreamingEventProcessor.Builder initialSegmentCount(int initialSegmentCount)

Sets the initial segment count used to create segments on start up. Only used whenever there are not segments stored in the configured TokenStore upon start up of this StreamingEventProcessor. The given value should at least be 1. Defaults to 16.

Parameters:

initialSegmentCount - an int specifying the initial segment count used to create segments on start up

Returns:

the current Builder instance, for fluent interfacing

initialToken

public PooledStreamingEventProcessor.Builder initialToken(@Nonnull
 Function<StreamableMessageSource<TrackedEventMessage<?>>,TrackingToken> initialToken)

Specifies the Function used to generate the initial TrackingTokens. The function will be given the configured StreamableMessageSource' so that its methods can be invoked for token creation.

Defaults to an automatic replay since the start of the stream.

More specifically, it defaults to a ReplayToken that starts streaming from the tail with the replay flag enabled until the head at the moment of initialization is reached.

Parameters:

initialToken - a Function generating the initial TrackingToken based on a given StreamableMessageSource

Returns:

the current Builder instance, for fluent interfacing

tokenClaimInterval

public PooledStreamingEventProcessor.Builder tokenClaimInterval(long tokenClaimInterval)

Specifies the time in milliseconds the processor's coordinator should wait after a failed attempt to claim any segments for processing. Generally, this means all segments are claimed. Defaults to 5000 milliseconds.

Parameters:

tokenClaimInterval - the time in milliseconds the processor's coordinator should wait after a failed attempt to claim any segments for processing

Returns:

the current Builder instance, for fluent interfacing

maxClaimedSegments

public PooledStreamingEventProcessor.Builder maxClaimedSegments(int maxClaimedSegments)

Sets the maximum number of segments this instance may claim.

Parameters:

maxClaimedSegments - The maximum number of segments this instance may claim.

Returns:

The current Builder instance, for fluent interfacing.

maxSegmentProvider

public PooledStreamingEventProcessor.Builder maxSegmentProvider(MaxSegmentProvider maxSegmentProvider)

Defines the maximum number of segment this StreamingEventProcessor may claim per instance. Defaults to MaxSegmentProvider.maxShort().

Parameters:

maxSegmentProvider - A MaxSegmentProvider providing the maximum number segments this StreamingEventProcessor may claim per instance.

Returns:

The current Builder instance, for fluent interfacing.

claimExtensionThreshold

public PooledStreamingEventProcessor.Builder claimExtensionThreshold(long claimExtensionThreshold)

Specifies a time in milliseconds the work packages of this processor should extend the claim on a TrackingToken. The threshold will only be met in absence of regular event processing, since that updates the TrackingToken automatically. Defaults to 5000 milliseconds.

Parameters:

claimExtensionThreshold - a time in milliseconds the work packages of this processor should extend the claim on a TrackingToken.

Returns:

the current Builder instance, for fluent interfacing

batchSize

public PooledStreamingEventProcessor.Builder batchSize(int batchSize)

Specifies the number of events to be processed inside a single transaction. Defaults to a batch size of 1.

Increasing this value with increase the processing speed dramatically, but requires certainty that the operations performed during event handling can be rolled back.

Parameters:

batchSize - the number of events to be processed inside a single transaction

Returns:

the current Builder instance, for fluent interfacing

clock

public PooledStreamingEventProcessor.Builder clock(@Nonnull
 Clock clock)

Defines the Clock used for time dependent operation by this EventProcessor. Used by the Coordinator and WorkPackage threads to decide when to perform certain tasks, like updating TrackingToken claims or when to unmark a Segment as "unclaimable". Defaults to GenericEventMessage.clock.

Parameters:

clock - the Clock used for time dependent operation by this EventProcessor

Returns:

the current Builder instance, for fluent interfacing

enableCoordinatorClaimExtension

public PooledStreamingEventProcessor.Builder enableCoordinatorClaimExtension()

Enables the Coordinator to extend the claims of its WorkPackages.

Enabling "coordinator claim extension" is an optimization as it relieves this effort from the WorkPackage. Toggling this feature may be particularly useful whenever the event handling task of the WorkPackage is lengthy. Either because of a hefty event handling component or because of a large batchSize(int).

An example of a lengthy processing tasks is whenever handling a batch of events exceeds half the claimTimeout of the TokenStore. The claimTimeout defaults to 10 seconds for all durable TokenStore implementations.

In both scenarios, there's a window of opportunity that the WorkPackage is not fast enough in extending the claim itself. Not being able to do so potentially causes token stealing by other instances of this PooledStreamingEventProcessor, thus overburdening the overall event processing task.

Note that enabling this feature will result in more frequent invocation of the TokenStore to update the tokens.

Returns:

The current Builder instance, for fluent interfacing.

build

public PooledStreamingEventProcessor build()

Initializes a PooledStreamingEventProcessor as specified through this Builder.

Returns:

a PooledStreamingEventProcessor as specified through this Builder

validate

protected void validate()
                 throws AxonConfigurationException

Description copied from class: AbstractEventProcessor.Builder

Validates whether the fields contained in this Builder are set accordingly.

Overrides:

validate in class AbstractEventProcessor.Builder

Throws:

AxonConfigurationException - if one field is asserted to be incorrect according to the Builder's specifications

name

public String name()

Returns the name of this PooledStreamingEventProcessor.

Returns:

the name of this PooledStreamingEventProcessor