Package org.axonframework.messaging.eventhandling.deadletter

Class CachingSequencedDeadLetterQueue<M extends Message>

java.lang.Object

org.axonframework.messaging.eventhandling.deadletter.CachingSequencedDeadLetterQueue<M>

Type Parameters:

M - The type of Message contained in the dead letters within this queue.

All Implemented Interfaces:

SequencedDeadLetterQueue<M>

@Internal
public class CachingSequencedDeadLetterQueue<M extends Message>
extends Object
implements SequencedDeadLetterQueue<M>

A decorator for SequencedDeadLetterQueue that adds per-segment caching of sequence identifiers to optimize contains(Object, ProcessingContext) lookups. This is particularly important for high-throughput event processing where checking if an event's sequence is already dead-lettered should be as fast as possible.

Each Segment gets its own independent SequenceIdentifierCache, obtained from the ProcessingContext. When a segment is released, only that segment's cache is removed via invalidateCache(ProcessingContext), leaving other segments' caches intact.

If no segment is available in the ProcessingContext (or the context is null), operations delegate directly to the underlying queue without caching.

Thread-safety note: This class is not thread-safe when performing operations for the same sequence identifier concurrently. It is designed for internal use by DeadLetteringEventHandlingComponent, where operations on a given sequence identifier are already serialized by the upstream SequencingEventHandlingComponent. Different segments can operate concurrently without interference. External synchronization must be provided if this class is used in other contexts where concurrent access to the same sequence identifier is possible.

Since:

5.1.0

Author:

Mateusz Nowak

See Also:

• SequenceIdentifierCache
• SequencedDeadLetterQueue

Constructor Summary

Constructors

Constructor	Description
CachingSequencedDeadLetterQueue(SequencedDeadLetterQueue<M> delegate)	Constructs a caching decorator with the given delegate and default cache settings.
CachingSequencedDeadLetterQueue(SequencedDeadLetterQueue<M> delegate, int cacheMaxSize)	Constructs a caching decorator with the given delegate and cache max size.

Method Summary

Modifier and Type	Method	Description
CompletableFuture<Long>	amountOfSequences(@Nullable ProcessingContext context)	Returns the number of unique sequences contained in this queue.
int	cacheEnqueuedSize()	Returns the total size of identifiers cached as enqueued, aggregated across all segments.
int	cacheNonEnqueuedSize()	Returns the total size of identifiers cached as not enqueued, aggregated across all segments.
CompletableFuture<Void>	clear(@Nullable ProcessingContext context)	Clears out all dead letters present in this queue.
CompletableFuture<Boolean>	contains(Object sequenceIdentifier, @Nullable ProcessingContext context)	Check whether there's a sequence of dead letters for the given sequenceIdentifier.
CompletableFuture<Iterable<Iterable<DeadLetter<? extends M>>>>	deadLetters(@Nullable ProcessingContext context)	Return all dead letter sequences held by this queue.
CompletableFuture<Iterable<DeadLetter<? extends M>>>	deadLetterSequence(Object sequenceIdentifier, @Nullable ProcessingContext context)	Return all the dead letters for the given sequenceIdentifier in insert order.
CompletableFuture<Void>	enqueue(Object sequenceIdentifier, DeadLetter<? extends M> letter, @Nullable ProcessingContext context)	Enqueues a dead letter containing an implementation of M to this queue.
CompletableFuture<Boolean>	enqueueIfPresent(Object sequenceIdentifier, Supplier<DeadLetter<? extends M>> letterBuilder, @Nullable ProcessingContext context)	Enqueue the result of the given letterBuilder only if there already are other dead letters with the same sequenceIdentifier present in this queue.
CompletableFuture<Void>	evict(DeadLetter<? extends M> letter, @Nullable ProcessingContext context)	Evict the given letter from this queue.
void	invalidateCache(@Nullable ProcessingContext context)	Invalidates the sequence identifier cache for the segment found in the given ProcessingContext.
CompletableFuture<Boolean>	isFull(Object sequenceIdentifier, @Nullable ProcessingContext context)	Validates whether this queue is full for the given sequenceIdentifier.
CompletableFuture<Boolean>	process(Predicate<DeadLetter<? extends M>> sequenceFilter, Function<DeadLetter<? extends M>,CompletableFuture<EnqueueDecision<M>>> processingTask, @Nullable ProcessingContext context)	Process a single sequence of enqueued dead letters through the given processingTask matching the sequenceFilter.
CompletableFuture<Void>	requeue(DeadLetter<? extends M> letter, UnaryOperator<DeadLetter<? extends M>> letterUpdater, @Nullable ProcessingContext context)	Reenters the given letter, updating the contents with the letterUpdater.
CompletableFuture<Long>	sequenceSize(Object sequenceIdentifier, @Nullable ProcessingContext context)	Returns the number of dead letters for the sequence matching the given sequenceIdentifier contained in this queue.
CompletableFuture<Long>	size(@Nullable ProcessingContext context)	Returns the number of dead letters contained in this queue.

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.axonframework.messaging.deadletter.SequencedDeadLetterQueue

process

Constructor Details

CachingSequencedDeadLetterQueue

public CachingSequencedDeadLetterQueue(SequencedDeadLetterQueue<M> delegate)

Constructs a caching decorator with the given delegate and default cache settings.

Each segment's cache is lazily initialized on first use by checking if the delegate queue is empty. If empty, the cache operates in an optimized mode where unknown identifiers are assumed to not be present.

Parameters:

delegate - the underlying SequencedDeadLetterQueue to delegate to

CachingSequencedDeadLetterQueue

public CachingSequencedDeadLetterQueue(SequencedDeadLetterQueue<M> delegate,
 int cacheMaxSize)

Constructs a caching decorator with the given delegate and cache max size.

Each segment's cache is lazily initialized on first use by checking if the delegate queue is empty. If empty, the cache operates in an optimized mode where unknown identifiers are assumed to not be present.

Parameters:

delegate - the underlying SequencedDeadLetterQueue to delegate to

cacheMaxSize - the maximum size of the non-enqueued identifiers cache

Method Details

enqueue

public CompletableFuture<Void> enqueue(Object sequenceIdentifier,
 DeadLetter<? extends M> letter,
 @Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Enqueues a dead letter containing an implementation of M to this queue.

The dead letter will be appended to a sequence depending on the sequenceIdentifier. If there is no sequence yet, it will construct one.

Specified by:

enqueue in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

sequenceIdentifier - The identifier of the sequence the letter belongs to.

letter - The DeadLetter to enqueue.

context - the processing context within which to enqueue the given letter

Returns:

A CompletableFuture that completes when the operation is done. The future completes exceptionally with a DeadLetterQueueOverflowException when this queue is full.

enqueueIfPresent

public CompletableFuture<Boolean> enqueueIfPresent(Object sequenceIdentifier,
 Supplier<DeadLetter<? extends M>> letterBuilder,
 @Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Enqueue the result of the given letterBuilder only if there already are other dead letters with the same sequenceIdentifier present in this queue.

Specified by:

enqueueIfPresent in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

sequenceIdentifier - The identifier of the sequence to store the result of the letterBuilder in.

letterBuilder - The DeadLetter builder constructing the letter to enqueue. Only invoked if the given sequenceIdentifier is contained.

context - the processing context within which to enqueue the result of the given letterBuilder if there is a sequence for the given sequenceIdentifier

Returns:

A CompletableFuture with true if there are dead letters for the given sequenceIdentifier and thus the letterBuilder's outcome is inserted. Otherwise, the future completes with false. The future completes exceptionally with a DeadLetterQueueOverflowException when this queue is SequencedDeadLetterQueue.isFull(Object, ProcessingContext) for the given sequenceIdentifier.

evict

public CompletableFuture<Void> evict(DeadLetter<? extends M> letter,
 @Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Evict the given letter from this queue. Nothing happens if the dead letter does not exist in this queue.

Specified by:

evict in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

letter - The dead letter to evict from this queue.

context - the processing context within which to evict the given letter

Returns:

A CompletableFuture that completes when the eviction is done.

requeue

public CompletableFuture<Void> requeue(DeadLetter<? extends M> letter,
 UnaryOperator<DeadLetter<? extends M>> letterUpdater,
 @Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Reenters the given letter, updating the contents with the letterUpdater. This method should be invoked if processing decided to keep the letter in the queue.

This operation adjusts the DeadLetter.lastTouched(). It may adjust the DeadLetter.cause() and DeadLetter.diagnostics(), depending on the given letterUpdater.

Specified by:

requeue in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

letter - The dead letter to reenter in this queue.

letterUpdater - A lambda taking in the given letter and updating the entry for requeueing. This may adjust the DeadLetter.cause() and DeadLetter.diagnostics(), for example.

context - the processing context within which to requeue the given letter

Returns:

A CompletableFuture that completes when the requeue is done. The future completes exceptionally with a NoSuchDeadLetterException if the given letter does not exist in the queue.

contains

public CompletableFuture<Boolean> contains(Object sequenceIdentifier,
 @Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Check whether there's a sequence of dead letters for the given sequenceIdentifier.

Specified by:

contains in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

sequenceIdentifier - The identifier used to validate for contained dead letters instances.

context - the ProcessingContext wherein this operation is invoked, if any

Returns:

A CompletableFuture with true if there are dead letters present for the given sequenceIdentifier, false otherwise.

deadLetterSequence

public CompletableFuture<Iterable<DeadLetter<? extends M>>> deadLetterSequence(Object sequenceIdentifier,
 @Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Return all the dead letters for the given sequenceIdentifier in insert order.

Specified by:

deadLetterSequence in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

sequenceIdentifier - The identifier of the sequence of dead letters to return.

context - the ProcessingContext wherein this operation is invoked, if any

Returns:

A CompletableFuture with all the dead letters for the given sequenceIdentifier in insert order.

deadLetters

public CompletableFuture<Iterable<Iterable<DeadLetter<? extends M>>>> deadLetters(@Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Return all dead letter sequences held by this queue. The sequences are not necessarily returned in insert order.

Specified by:

deadLetters in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

context - the ProcessingContext wherein this operation is invoked, if any

Returns:

A CompletableFuture with all dead letter sequences held by this queue.

isFull

public CompletableFuture<Boolean> isFull(Object sequenceIdentifier,
 @Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Validates whether this queue is full for the given sequenceIdentifier.

This method returns true either when the maximum amount of sequences or the maximum sequence size is reached.

Specified by:

isFull in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

sequenceIdentifier - The identifier of the sequence to validate for.

context - the ProcessingContext wherein this operation is invoked, if any

Returns:

A CompletableFuture with true either when the limit of this queue is reached. Contains false otherwise.

size

public CompletableFuture<Long> size(@Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Returns the number of dead letters contained in this queue.

Specified by:

size in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

context - the ProcessingContext wherein this operation is invoked, if any

Returns:

A CompletableFuture with the number of dead letters contained in this queue.

sequenceSize

public CompletableFuture<Long> sequenceSize(Object sequenceIdentifier,
 @Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Returns the number of dead letters for the sequence matching the given sequenceIdentifier contained in this queue.

Note that there's a window of opportunity where the size might exceed the maximum sequence size to account for concurrent usage.

Specified by:

sequenceSize in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

sequenceIdentifier - The identifier of the sequence to retrieve the size from.

context - the ProcessingContext wherein this operation is invoked, if any

Returns:

A CompletableFuture with the number of dead letters for the sequence matching the given sequenceIdentifier.

amountOfSequences

public CompletableFuture<Long> amountOfSequences(@Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Returns the number of unique sequences contained in this queue.

Note that there's a window of opportunity where the size might exceed the maximum amount of sequences to account for concurrent usage of this dead letter queue.

Specified by:

amountOfSequences in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

context - the ProcessingContext wherein this operation is invoked, if any

Returns:

A CompletableFuture with the number of unique sequences contained in this queue.

process

public CompletableFuture<Boolean> process(Predicate<DeadLetter<? extends M>> sequenceFilter,
 Function<DeadLetter<? extends M>,CompletableFuture<EnqueueDecision<M>>> processingTask,
 @Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Process a single sequence of enqueued dead letters through the given processingTask matching the sequenceFilter. It will pick the oldest available sequence, determined by the DeadLetter.lastTouched() field of the first entry in each sequence.

Note that only a single matching sequence is processed! Furthermore, only the first dead letter is validated, because it is the blocker for the processing of the rest of the sequence.

Uses the EnqueueDecision returned by the processingTask to decide whether to SequencedDeadLetterQueue.evict(DeadLetter, ProcessingContext) or SequencedDeadLetterQueue.requeue(DeadLetter, UnaryOperator, ProcessingContext) a dead letter from the selected sequence. The processingTask is invoked as long as letters are present in the selected sequence and the result of processing returns false for EnqueueDecision.shouldEnqueue() decision. The latter means the dead letter should be evicted.

This operation protects against concurrent invocations of the processingTask on the filtered sequence. Doing so ensures enqueued messages are handled in order.

Specified by:

process in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

sequenceFilter - A lambda selecting the sequences within this queue to process with the processingTask.

processingTask - A function processing a dead letter. Returns a CompletableFuture with an EnqueueDecision used to deduce whether to SequencedDeadLetterQueue.evict(DeadLetter, ProcessingContext) or SequencedDeadLetterQueue.requeue(DeadLetter, UnaryOperator, ProcessingContext) the dead letter.

context - the ProcessingContext in which the dead letters are processed

Returns:

A CompletableFuture with true if an entire sequence of dead letters was processed successfully, false otherwise. This means the processingTask processed all dead letters of a sequence and the outcome was to evict each instance.

clear

public CompletableFuture<Void> clear(@Nullable ProcessingContext context)

Clears out all dead letters present in this queue.

Clears all per-segment caches and the delegate queue.

Specified by:

clear in interface SequencedDeadLetterQueue<M extends Message>

Parameters:

context - the ProcessingContext wherein this operation is invoked, if any

Returns:

A CompletableFuture that completes when all dead letters have been cleared.

invalidateCache

public void invalidateCache(@Nullable ProcessingContext context)

Invalidates the sequence identifier cache for the segment found in the given ProcessingContext.

The segment is extracted from the context internally. If no segment is present in the context, this method is a no-op.

Call this method when processing ownership changes (e.g., segment release) to ensure cache consistency. When ownership changes, another processor instance may have modified the queue, making cached information potentially stale.

Parameters:

context - the processing context containing the Segment to invalidate the cache for, may be null

cacheEnqueuedSize

public int cacheEnqueuedSize()

Returns the total size of identifiers cached as enqueued, aggregated across all segments.

Returns:

the total count of enqueued identifiers across all segment caches

cacheNonEnqueuedSize

public int cacheNonEnqueuedSize()

Returns the total size of identifiers cached as not enqueued, aggregated across all segments.

Returns:

the total count of non-enqueued identifiers across all segment caches