public class PooledStreamingEventProcessor extends AbstractEventProcessor implements StreamingEventProcessor, Lifecycle
StreamingEventProcessor implementation which pools its 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).
| Modifier and Type | Class and Description |
|---|---|
static class |
PooledStreamingEventProcessor.Builder
Builder class to instantiate a
PooledStreamingEventProcessor. |
Lifecycle.LifecycleHandler, Lifecycle.LifecycleRegistryspanFactory| Modifier | Constructor and Description |
|---|---|
protected |
PooledStreamingEventProcessor(PooledStreamingEventProcessor.Builder builder)
Instantiate a
PooledStreamingEventProcessor based on the fields contained in the PooledStreamingEventProcessor.Builder. |
| Modifier and Type | Method and Description |
|---|---|
static PooledStreamingEventProcessor.Builder |
builder()
Instantiate a Builder to be able to create a
PooledStreamingEventProcessor. |
CompletableFuture<Boolean> |
claimSegment(int segmentId)
Instructs the processor to claim the segment with given
segmentId and start processing it as soon as
possible. |
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". |
canHandle, canHandleType, eventHandlerInvoker, getHandlerInterceptors, getName, processInUnitOfWork, processInUnitOfWork, registerHandlerInterceptor, reportIgnored, toStringclone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, waitisReplayinggetHandlerInterceptors, getNameregisterHandlerInterceptorprotected PooledStreamingEventProcessor(PooledStreamingEventProcessor.Builder builder)
PooledStreamingEventProcessor based on the fields contained in the PooledStreamingEventProcessor.Builder.
Will assert the following for their presence prior to constructing this processor:
name.EventHandlerInvoker.StreamableMessageSource.TokenStore.TransactionManager.ScheduledExecutorService for coordination.ScheduledExecutorService to process work packages.AxonConfigurationException is thrown.builder - the PooledStreamingEventProcessor.Builder used to instantiate a PooledStreamingEventProcessor instancepublic static PooledStreamingEventProcessor.Builder builder()
PooledStreamingEventProcessor.
Upon initialization of this builder, the following fields are defaulted:
RollbackConfigurationType defaults to a RollbackConfigurationType.ANY_THROWABLE.ErrorHandler is defaulted to a PropagatingErrorHandler.MessageMonitor defaults to a NoOpMessageMonitor.initialSegmentCount defaults to 16.initialToken function defaults to StreamableMessageSource.createTailToken().tokenClaimInterval defaults to 5000 milliseconds.MaxSegmentProvider (used by maxCapacity()) defaults to MaxSegmentProvider.maxShort().claimExtensionThreshold defaults to 5000 milliseconds.batchSize defaults to 1.Clock defaults to GenericEventMessage.clock.EventProcessorSpanFactory defaults to a DefaultEventProcessorSpanFactory backed by a NoOpSpanFactory.coordinatorExtendsClaims defaults to a false.EventProcessor.EventHandlerInvoker which will be given the events handled by this processorStreamableMessageSource used to retrieve events.TokenStore to store the progress of this processor in.ScheduledExecutorService to coordinate events and segment operations.ScheduledExecutorService to process work packages.PooledStreamingEventProcessorpublic void registerLifecycleHandlers(@Nonnull Lifecycle.LifecycleRegistry handle)
LifecycleregisterLifecycleHandlers in interface Lifecyclehandle - the lifecycle instance to register the handlers withLifecycle.LifecycleRegistry.onShutdown(int, Runnable),
LifecycleRegistry#onShutdown(int, LifecycleHandler),
Lifecycle.LifecycleRegistry.onStart(int, Runnable),
LifecycleRegistry#onStart(int, LifecycleHandler)public void start()
EventProcessorstart in interface EventProcessorpublic void shutDown()
EventProcessorshutDown in interface EventProcessorpublic CompletableFuture<Void> shutdownAsync()
EventProcessorCompletableFuture that completes when the shutdown process is
finished.shutdownAsync in interface EventProcessorpublic boolean isRunning()
EventProcessorisRunning in interface EventProcessortrue when running, otherwise falsepublic boolean isError()
EventProcessor
Note that this method returns false when the processor was stopped using EventProcessor.shutDown().
isError in interface EventProcessortrue when paused due to an error, otherwise falsepublic String getTokenStoreIdentifier()
StreamingEventProcessorTokenStore used by this StreamingEventProcessor.getTokenStoreIdentifier in interface StreamingEventProcessorTokenStore used by this StreamingEventProcessorpublic void releaseSegment(int segmentId)
StreamingEventProcessorsegmentId.releaseSegment in interface StreamingEventProcessorsegmentId - the id of the segment to releasepublic void releaseSegment(int segmentId,
long releaseDuration,
TimeUnit unit)
StreamingEventProcessorsegmentId. 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.
releaseSegment in interface StreamingEventProcessorsegmentId - the id of the segment to be released for the specified releaseDurationreleaseDuration - the amount of time to disregard segmentId for processingunit - the unit of time used to express the releaseDurationpublic CompletableFuture<Boolean> claimSegment(int segmentId)
StreamingEventProcessorsegmentId 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.
claimSegment in interface StreamingEventProcessorsegmentId - the identifier of the segment to claim and start processingCompletableFuture providing the result of the claim operationpublic CompletableFuture<Boolean> splitSegment(int segmentId)
StreamingEventProcessorsegmentId 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.
splitSegment in interface StreamingEventProcessorsegmentId - the identifier of the segment to splitCompletableFuture providing the result of the split operationpublic CompletableFuture<Boolean> mergeSegment(int segmentId)
StreamingEventProcessorsegmentId 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();
mergeSegment in interface StreamingEventProcessorsegmentId - the identifier of the segment to mergeCompletableFuture indicating whether the merge was executed successfullypublic boolean supportsReset()
StreamingEventProcessorStreamingEventProcessor 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.
supportsReset in interface StreamingEventProcessortrue if resets are supported, false otherwisepublic void resetTokens()
StreamingEventProcessorBefore 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 in interface StreamingEventProcessorpublic <R> void resetTokens(R resetContext)
StreamingEventProcessorresetContext 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.
resetTokens in interface StreamingEventProcessorR - the type of the provided resetContextresetContext - a R used to support the reset operationpublic void resetTokens(@Nonnull Function<StreamableMessageSource<TrackedEventMessage<?>>,TrackingToken> initialTrackingTokenSupplier)
StreamingEventProcessorinitialTrackingTokenSupplier. 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.
resetTokens in interface StreamingEventProcessorinitialTrackingTokenSupplier - a function returning the token representing the position to reset topublic <R> void resetTokens(@Nonnull Function<StreamableMessageSource<TrackedEventMessage<?>>,TrackingToken> initialTrackingTokenSupplier, R resetContext)
StreamingEventProcessorinitialTrackingTokenSupplier. 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.
resetTokens in interface StreamingEventProcessorR - the type of the provided resetContextinitialTrackingTokenSupplier - a function returning the token representing the position to reset toresetContext - a R used to support the reset operationpublic void resetTokens(@Nonnull TrackingToken startPosition)
StreamingEventProcessorstartPosition. 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.
resetTokens in interface StreamingEventProcessorstartPosition - the token representing the position to reset the processor topublic <R> void resetTokens(@Nonnull TrackingToken startPosition, R resetContext)
StreamingEventProcessorstartPosition. 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.
resetTokens in interface StreamingEventProcessorR - the type of the provided resetContextstartPosition - the token representing the position to reset the processor toresetContext - a R used to support the reset operationpublic int maxCapacity()
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.
maxCapacity in interface StreamingEventProcessorEventProcessor can process at the same timepublic Map<Integer,EventTrackerStatus> processingStatus()
StreamingEventProcessorEventTrackerStatus 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.
processingStatus in interface StreamingEventProcessorCopyright © 2010–2024. All rights reserved.