Package org.axonframework.messaging.eventhandling.processing.streaming.pooled

Class PooledStreamingEventProcessorConfiguration

java.lang.Object

org.axonframework.messaging.eventhandling.configuration.EventProcessorConfiguration

org.axonframework.messaging.eventhandling.processing.streaming.pooled.PooledStreamingEventProcessorConfiguration

All Implemented Interfaces:

DescribableComponent

public class PooledStreamingEventProcessorConfiguration
extends EventProcessorConfiguration

Configuration class for a PooledStreamingEventProcessor.

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

• The ErrorHandler is defaulted to a PropagatingErrorHandler.
• 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 coordinatorExtendsClaims defaults to a false.

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

• A StreamableEventSource 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.

Since:

5.0.0

Author:

Mateusz Nowak

Field Summary

Fields inherited from class org.axonframework.messaging.eventhandling.configuration.EventProcessorConfiguration

errorHandler, interceptors, messageMonitor, unitOfWorkFactory

Constructor Summary

Constructors

Constructor	Description
PooledStreamingEventProcessorConfiguration()	Constructs a new PooledStreamingEventProcessorConfiguration with just default values.
PooledStreamingEventProcessorConfiguration(EventProcessorConfiguration base)	Constructs a new PooledStreamingEventProcessorConfiguration copying properties from the given configuration.
PooledStreamingEventProcessorConfiguration(EventProcessorConfiguration base, Configuration configuration)	Constructs a new PooledStreamingEventProcessorConfiguration with default values and retrieve global default values.

Method Summary

Modifier and Type	Method	Description
int	batchSize()	Returns the number of events processed in a single transaction.
PooledStreamingEventProcessorConfiguration	batchSize(int batchSize)	Specifies the number of events to be processed inside a single transaction.
long	claimExtensionThreshold()	Returns the claim extension threshold in milliseconds.
PooledStreamingEventProcessorConfiguration	claimExtensionThreshold(long claimExtensionThreshold)	Specifies a time in milliseconds the work packages of this processor should extend the claim on a TrackingToken.
Clock	clock()	Returns the Clock used for time-dependent operations.
PooledStreamingEventProcessorConfiguration	clock(Clock clock)	Defines the Clock used for time dependent operation by this EventProcessor.
ScheduledExecutorService	coordinatorExecutor()	Returns the coordinator's ScheduledExecutorService.
PooledStreamingEventProcessorConfiguration	coordinatorExecutor(ScheduledExecutorService coordinatorExecutor)	Specifies the ScheduledExecutorService used by the coordinator of this PooledStreamingEventProcessor.
PooledStreamingEventProcessorConfiguration	coordinatorExecutor(Supplier<ScheduledExecutorService> coordinatorExecutor)	Specifies the ScheduledExecutorService used by the coordinator of this PooledStreamingEventProcessor.
boolean	coordinatorExtendsClaims()	Returns whether the coordinator extends claims for work packages.
void	describeTo(ComponentDescriptor descriptor)	Describe the properties of this DescribableComponent with the given descriptor.
PooledStreamingEventProcessorConfiguration	enableCoordinatorClaimExtension()	Enables the Coordinator to extend the claims of its WorkPackages.
PooledStreamingEventProcessorConfiguration	errorHandler(ErrorHandler errorHandler)	Sets the ErrorHandler invoked when an UnitOfWork throws an exception during processing.
PooledStreamingEventProcessorConfiguration	eventCriteria(Function<Set<QualifiedName>,EventCriteria> eventCriteriaProvider)	Sets the function to build the EventCriteria used to filter events when opening the event source.
Function<Set<QualifiedName>,EventCriteria>	eventCriteriaProvider()	Returns the function to build EventCriteria from supported event types.
StreamableEventSource	eventSource()	Returns the StreamableEventSource used to track events.
PooledStreamingEventProcessorConfiguration	eventSource(StreamableEventSource eventSource)	Sets the StreamableEventSource (e.g. the EventStore) which this EventProcessor will track.
Consumer<? super EventMessage>	ignoredMessageHandler()	Returns the handler for ignored messages.
PooledStreamingEventProcessorConfiguration	ignoredMessageHandler(Consumer<? super EventMessage> ignoredMessageHandler)	Sets the handler, that is invoked when the event is ignored by all WorkPackages this Coordinator controls.
int	initialSegmentCount()	Returns the initial segment count used on startup.
PooledStreamingEventProcessorConfiguration	initialSegmentCount(int initialSegmentCount)	Sets the initial segment count used to create segments on start up.
Function<TrackingTokenSource,CompletableFuture<TrackingToken>>	initialToken()	Returns the function used to generate initial TrackingTokens.
PooledStreamingEventProcessorConfiguration	initialToken(Function<TrackingTokenSource,CompletableFuture<TrackingToken>> initialToken)	Specifies the Function used to generate the initial TrackingTokens.
PooledStreamingEventProcessorConfiguration	maxClaimedSegments(int maxClaimedSegments)	Sets the maximum number of segments this instance may claim.
MaxSegmentProvider	maxSegmentProvider()	Returns the MaxSegmentProvider defining maximum claimable segments.
PooledStreamingEventProcessorConfiguration	maxSegmentProvider(MaxSegmentProvider maxSegmentProvider)	Defines the maximum number of segment this StreamingEventProcessor may claim per instance.
Supplier<ProcessingContext>	schedulingProcessingContextProvider()	Returns the Supplier providing the ProcessingContext used to evaluate whether an event can be scheduled for processing by this WorkPackage.
PooledStreamingEventProcessorConfiguration	schedulingProcessingContextProvider(Supplier<ProcessingContext> schedulingProcessingContextProvider)	Provides a ProcessingContext used to evaluate whether an event can be scheduled for processing by this WorkPackage.
boolean	streaming()	Returns whether this configuration is for a streaming event processor.
long	tokenClaimInterval()	Returns the token claim interval in milliseconds.
PooledStreamingEventProcessorConfiguration	tokenClaimInterval(long tokenClaimInterval)	Specifies the time in milliseconds the processor's coordinator should wait after a failed attempt to claim any segments for processing.
TokenStore	tokenStore()	Returns the TokenStore used to store and fetch event tokens.
PooledStreamingEventProcessorConfiguration	tokenStore(TokenStore tokenStore)	Sets the TokenStore used to store and fetch event tokens that enable this EventProcessor to track its progress.
PooledStreamingEventProcessorConfiguration	unitOfWorkFactory(UnitOfWorkFactory unitOfWorkFactory)	A UnitOfWorkFactory that spawns UnitOfWork used to process an event batch.
protected void	validate()	Validates whether the fields contained in this Builder are set accordingly.
PooledStreamingEventProcessorConfiguration	withInterceptor(MessageHandlerInterceptor<? super EventMessage> interceptor)	Registers the given EventMessage-specific MessageHandlerInterceptor for the PooledStreamingEventProcessor under construction.
ScheduledExecutorService	workerExecutor()	Returns the worker's ScheduledExecutorService.
PooledStreamingEventProcessorConfiguration	workerExecutor(ScheduledExecutorService workerExecutor)	Specifies the ScheduledExecutorService to be provided to the WorkPackages created by this PooledStreamingEventProcessor.
PooledStreamingEventProcessorConfiguration	workerExecutor(Supplier<ScheduledExecutorService> workerExecutor)	Specifies the ScheduledExecutorService to be provided to the WorkPackages created by this PooledStreamingEventProcessor.

Methods inherited from class org.axonframework.messaging.eventhandling.configuration.EventProcessorConfiguration

errorHandler, interceptors, unitOfWorkFactory

Methods inherited from class java.lang.Object

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

Constructor Details

PooledStreamingEventProcessorConfiguration

@Internal
public PooledStreamingEventProcessorConfiguration()

Constructs a new PooledStreamingEventProcessorConfiguration with just default values. Do not retrieve any global default values.

This configuration will not have any of the default MessageHandlerInterceptors for events. Please use PooledStreamingEventProcessorConfiguration(EventProcessorConfiguration, Configuration) when those are desired.

PooledStreamingEventProcessorConfiguration

@Internal
public PooledStreamingEventProcessorConfiguration(@Nonnull
 EventProcessorConfiguration base)

Constructs a new PooledStreamingEventProcessorConfiguration copying properties from the given configuration.

Parameters:

base - The EventProcessorConfiguration to copy properties from.

PooledStreamingEventProcessorConfiguration

@Internal
public PooledStreamingEventProcessorConfiguration(@Nonnull
 EventProcessorConfiguration base,
 @Nonnull
 Configuration configuration)

Constructs a new PooledStreamingEventProcessorConfiguration with default values and retrieve global default values.

Parameters:

base - The EventProcessorConfiguration to copy properties from.

configuration - The configuration, used to retrieve global default values, like MessageHandlerInterceptors, from.

Method Details

errorHandler

public PooledStreamingEventProcessorConfiguration errorHandler(@Nonnull
 ErrorHandler errorHandler)

Description copied from class: EventProcessorConfiguration

Sets the ErrorHandler invoked when an UnitOfWork throws an exception during processing. Defaults to a PropagatingErrorHandler.

Overrides:

errorHandler in class EventProcessorConfiguration

Parameters:

errorHandler - the ErrorHandler invoked when an UnitOfWork throws an exception during processing

Returns:

The current instance, for fluent interfacing.

unitOfWorkFactory

public PooledStreamingEventProcessorConfiguration unitOfWorkFactory(@Nonnull
 UnitOfWorkFactory unitOfWorkFactory)

Description copied from class: EventProcessorConfiguration

A UnitOfWorkFactory that spawns UnitOfWork used to process an event batch.

Overrides:

unitOfWorkFactory in class EventProcessorConfiguration

Parameters:

unitOfWorkFactory - A UnitOfWorkFactory that spawns UnitOfWork.

Returns:

The current instance, for fluent interfacing.

eventSource

public PooledStreamingEventProcessorConfiguration eventSource(@Nonnull
 StreamableEventSource eventSource)

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

Parameters:

eventSource - The StreamableEventSource (e.g. the EventStore) which this EventProcessor will track.

Returns:

The current instance, for fluent interfacing.

withInterceptor

@Nonnull
public PooledStreamingEventProcessorConfiguration withInterceptor(@Nonnull
 MessageHandlerInterceptor<? super EventMessage> interceptor)

Registers the given EventMessage-specific MessageHandlerInterceptor for the PooledStreamingEventProcessor under construction.

Parameters:

interceptor - The EventMessage-specific MessageHandlerInterceptor to register for the PooledStreamingEventProcessor under construction.

Returns:

This PooledStreamingEventProcessorConfiguration, for fluent interfacing.

tokenStore

public PooledStreamingEventProcessorConfiguration 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 instance, for fluent interfacing.

coordinatorExecutor

public PooledStreamingEventProcessorConfiguration coordinatorExecutor(@Nonnull
 ScheduledExecutorService coordinatorExecutor)

Specifies the ScheduledExecutorService used by the coordinator of this PooledStreamingEventProcessor.

Parameters:

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

Returns:

The current instance, for fluent interfacing.

coordinatorExecutor

public PooledStreamingEventProcessorConfiguration coordinatorExecutor(@Nonnull
 Supplier<ScheduledExecutorService> coordinatorExecutor)

Specifies the ScheduledExecutorService used by the coordinator of this PooledStreamingEventProcessor.

Parameters:

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

Returns:

The current instance, for fluent interfacing.

workerExecutor

public PooledStreamingEventProcessorConfiguration workerExecutor(@Nonnull
 ScheduledExecutorService workerExecutor)

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

Note that workerExecutor(Supplier) is favored over this method, as it avoids eager initialization of an executor that may be overridden by other components.

Parameters:

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

Returns:

The current instance, for fluent interfacing.

workerExecutor

public PooledStreamingEventProcessorConfiguration workerExecutor(@Nonnull
 Supplier<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 instance, for fluent interfacing.

initialSegmentCount

public PooledStreamingEventProcessorConfiguration initialSegmentCount(int initialSegmentCount)

Sets the initial segment count used to create segments on start up. Only used whenever there are no 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 - The int specifying the initial segment count used to create segments on startup.

Returns:

The current instance, for fluent interfacing.

initialToken

public PooledStreamingEventProcessorConfiguration initialToken(@Nonnull
 Function<TrackingTokenSource,CompletableFuture<TrackingToken>> initialToken)

Specifies the Function used to generate the initial TrackingTokens. The function will be given the configured StreamableEventSource 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 - The Function generating the initial TrackingToken based on a given StreamableEventSource.

Returns:

The current instance, for fluent interfacing.

tokenClaimInterval

public PooledStreamingEventProcessorConfiguration 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 instance, for fluent interfacing.

maxClaimedSegments

public PooledStreamingEventProcessorConfiguration 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 instance, for fluent interfacing.

maxSegmentProvider

public PooledStreamingEventProcessorConfiguration maxSegmentProvider(@Nonnull
 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 instance, for fluent interfacing.

claimExtensionThreshold

public PooledStreamingEventProcessorConfiguration 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 update the TrackingToken automatically. Defaults to 5000 milliseconds.

Parameters:

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

Returns:

The current instance, for fluent interfacing

batchSize

public PooledStreamingEventProcessorConfiguration 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 instance, for fluent interfacing.

clock

public PooledStreamingEventProcessorConfiguration 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 instance, for fluent interfacing.

enableCoordinatorClaimExtension

public PooledStreamingEventProcessorConfiguration 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 instance, for fluent interfacing.

ignoredMessageHandler

public PooledStreamingEventProcessorConfiguration ignoredMessageHandler(Consumer<? super EventMessage> ignoredMessageHandler)

Sets the handler, that is invoked when the event is ignored by all WorkPackages this Coordinator controls. Defaults to a no-op.

Parameters:

ignoredMessageHandler - The handler, that is invoked when the event is ignored by all * WorkPackages this Coordinator controls.

Returns:

The current Builder instance, for fluent interfacing.

eventCriteria

public PooledStreamingEventProcessorConfiguration eventCriteria(@Nonnull
 Function<Set<QualifiedName>,EventCriteria> eventCriteriaProvider)

Sets the function to build the EventCriteria used to filter events when opening the event source. The function receives the set of supported event types from the assigned EventHandlingComponent.

Intention: This function is mainly intended to allow you to specify the tags for filtering or to build more complex criteria. For example, if not all supported event types share the same tag, you may use EventCriteria.either(EventCriteria...) to construct a disjunction of criteria for different event types and tags. See EventCriteria for advanced usage and examples.

By default, it returns EventCriteria.havingAnyTag().andBeingOneOfTypes(supportedEvents).

Parameters:

eventCriteriaProvider - The function to build the EventCriteria from supported event types.

Returns:

The current instance, for fluent interfacing.

schedulingProcessingContextProvider

public PooledStreamingEventProcessorConfiguration schedulingProcessingContextProvider(@Nonnull
 Supplier<ProcessingContext> schedulingProcessingContextProvider)

Provides a ProcessingContext used to evaluate whether an event can be scheduled for processing by this WorkPackage. The provided ProcessingContext is enriched with resources from the MessageStream.Entry to evaluate whether the event can be handled by this package's Segment. Currently, the only usage of the context is for EventHandlingComponent.sequenceIdentifierFor(EventMessage, ProcessingContext) execution.

Parameters:

schedulingProcessingContextProvider - The ProcessingContext provider.

Returns:

The current Builder instance, for fluent interfacing.

validate

protected void validate()
                 throws AxonConfigurationException

Description copied from class: EventProcessorConfiguration

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

Overrides:

validate in class EventProcessorConfiguration

Throws:

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

streaming

public boolean streaming()

Description copied from class: EventProcessorConfiguration

Returns whether this configuration is for a streaming event processor.

Overrides:

streaming in class EventProcessorConfiguration

Returns:

false for basic configuration, true for streaming configurations.

eventSource

public StreamableEventSource eventSource()

Returns the StreamableEventSource used to track events.

Returns:

The StreamableEventSource for this processor.

tokenStore

public TokenStore tokenStore()

Returns the TokenStore used to store and fetch event tokens.

Returns:

The TokenStore for tracking progress.

coordinatorExecutor

public ScheduledExecutorService coordinatorExecutor()

Returns the coordinator's ScheduledExecutorService.

Returns:

The coordinator executor.

workerExecutor

public ScheduledExecutorService workerExecutor()

Returns the worker's ScheduledExecutorService.

Returns:

The worker executor.

initialSegmentCount

public int initialSegmentCount()

Returns the initial segment count used on startup.

Returns:

The initial segment count.

initialToken

public Function<TrackingTokenSource,CompletableFuture<TrackingToken>> initialToken()

Returns the function used to generate initial TrackingTokens.

Returns:

The initial token generation function.

tokenClaimInterval

public long tokenClaimInterval()

Returns the token claim interval in milliseconds.

Returns:

The token claim interval.

maxSegmentProvider

public MaxSegmentProvider maxSegmentProvider()

Returns the MaxSegmentProvider defining maximum claimable segments.

Returns:

The MaxSegmentProvider for this processor.

claimExtensionThreshold

public long claimExtensionThreshold()

Returns the claim extension threshold in milliseconds.

Returns:

The claim extension threshold.

batchSize

public int batchSize()

Returns the number of events processed in a single transaction.

Returns:

The batch size.

clock

public Clock clock()

Returns the Clock used for time-dependent operations.

Returns:

The Clock for this processor.

coordinatorExtendsClaims

public boolean coordinatorExtendsClaims()

Returns whether the coordinator extends claims for work packages.

Returns:

true if coordinator extends claims, false otherwise.

eventCriteriaProvider

public Function<Set<QualifiedName>,EventCriteria> eventCriteriaProvider()

Returns the function to build EventCriteria from supported event types.

Returns:

The event criteria provider function.

ignoredMessageHandler

public Consumer<? super EventMessage> ignoredMessageHandler()

Returns the handler for ignored messages.

Returns:

The ignored message handler.

schedulingProcessingContextProvider

public Supplier<ProcessingContext> schedulingProcessingContextProvider()

Returns the Supplier providing the ProcessingContext used to evaluate whether an event can be scheduled for processing by this WorkPackage.

Returns:

The ProcessingContext provider.

describeTo

public void describeTo(@Nonnull
 ComponentDescriptor descriptor)

Description copied from interface: DescribableComponent

Describe the properties of this DescribableComponent with the given descriptor.

Components should call the appropriate describeProperty methods on the descriptor to register their properties. The descriptor is responsible for determining how these properties are formatted and structured in the final output.

Best Practices: As a general rule, all relevant fields of a DescribableComponent implementation should be described in this method. However, developers have discretion to include only the fields that make sense in the context. Not every field may be meaningful for description purposes, especially internal implementation details. Furthermore, components might want to expose different information based on their current state. The final decision on what properties to include lies with the person implementing the describeTo method, who should focus on providing information that is useful for understanding the component's configuration and state.

Example implementation:

 public void describeTo(ComponentDescriptor descriptor) {
     descriptor.describeProperty("name", this.name);
     descriptor.describeProperty("enabled", this.enabled);
     descriptor.describeProperty("configuration", this.configuration); // A nested component
     descriptor.describeProperty("handlers", this.eventHandlers);      // A collection
 }
 

Specified by:

describeTo in interface DescribableComponent

Overrides:

describeTo in class EventProcessorConfiguration

Parameters:

descriptor - The component descriptor to describe this DescribableComponentn its properties in.