Package org.axonframework.messaging.eventhandling.deadletter.jdbc

Class JdbcSequencedDeadLetterQueue<E extends EventMessage>

java.lang.Object

org.axonframework.messaging.eventhandling.deadletter.jdbc.JdbcSequencedDeadLetterQueue<E>

Type Parameters:

E - An implementation of EventMessage contained in the dead-letters within this queue.

All Implemented Interfaces:

SequencedDeadLetterQueue<E>

public class JdbcSequencedDeadLetterQueue<E extends EventMessage>
extends Object
implements SequencedDeadLetterQueue<E>

A JDBC-based implementation of the SequencedDeadLetterQueue, used for storing dead letters containing event messages durably. Use the

invalid reference

#createSchema(DeadLetterTableFactory)

operation to build the table and indices required by this SequencedDeadLetterQueue, providing the desired DeadLetterTableFactory. The statements used by this queues methods can be optimized by providing a custom DeadLetterStatementFactory.

Keeps the insertion order intact by saving an incremented index within each unique sequence, backed by the index property. Each sequence is uniquely identified by the sequence identifier, stored in the DeadLetterSchema.sequenceIdentifierColumn() sequence identifier} field.

When processing an item, single execution across all applications is guaranteed by setting the processing started property, locking other processes out of the sequence for the configured claimDuration (30 seconds by default).

The stored entries are converted to a JdbcDeadLetter when they need to be processed or filtered. In order to restore the original EventMessage the DeadLetterJdbcConverter is used. The default supports all EventMessage implementations provided by the framework. If you have a custom variant, you have to build your own.

invalid reference

upcasters

are not supported by this implementation, so breaking changes for events messages stored in the queue should be avoided.

Since:

4.8.0

Author:

Steven van Beelen

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	JdbcSequencedDeadLetterQueue.Builder<E extends EventMessage>	Builder class to instantiate a JdbcSequencedDeadLetterQueue.

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	JdbcSequencedDeadLetterQueue(JdbcSequencedDeadLetterQueue.Builder<E> builder)	Instantiate a JDBC-based SequencedDeadLetterQueue through the given builder.

Method Summary

Modifier and Type	Method	Description
@NonNull CompletableFuture<Long>	amountOfSequences(@Nullable ProcessingContext context)	Returns the number of unique sequences contained in this queue.
static <E extends EventMessage> JdbcSequencedDeadLetterQueue.Builder<E>	builder()	Instantiate a builder to construct a JdbcSequencedDeadLetterQueue.
@NonNull 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<Void>	createSchema(DeadLetterTableFactory tableFactory, @Nullable ProcessingContext context)	Performs the DDL queries to create the schema necessary for this SequencedDeadLetterQueue implementation.
@NonNull CompletableFuture<Iterable<Iterable<DeadLetter<? extends E>>>>	deadLetters(@Nullable ProcessingContext context)	Return all dead letter sequences held by this queue.
CompletableFuture<Iterable<DeadLetter<? extends E>>>	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 E> letter, @Nullable ProcessingContext context)	Enqueues a dead letter containing an implementation of M to this queue.
@NonNull CompletableFuture<Void>	evict(@NonNull DeadLetter<? extends E> letter, @Nullable ProcessingContext context)	Evict the given letter from this queue.
CompletableFuture<Boolean>	isFull(Object sequenceIdentifier, @Nullable ProcessingContext context)	Validates whether this queue is full for the given sequenceIdentifier.
@NonNull CompletableFuture<Boolean>	process(@NonNull Predicate<DeadLetter<? extends E>> sequenceFilter, @NonNull Function<DeadLetter<? extends E>,CompletableFuture<EnqueueDecision<E>>> processingTask, @Nullable ProcessingContext context)	Process a single sequence of enqueued dead letters through the given processingTask matching the sequenceFilter.
CompletableFuture<Boolean>	process(Function<DeadLetter<? extends E>,CompletableFuture<EnqueueDecision<E>>> processingTask, @Nullable ProcessingContext context)	Process a single sequence of enqueued dead letters with the given processingTask.
CompletableFuture<Void>	requeue(DeadLetter<? extends E> letter, UnaryOperator<DeadLetter<? extends E>> 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.
@NonNull 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

enqueueIfPresent

Constructor Details

JdbcSequencedDeadLetterQueue

protected JdbcSequencedDeadLetterQueue(JdbcSequencedDeadLetterQueue.Builder<E> builder)

Instantiate a JDBC-based SequencedDeadLetterQueue through the given builder.

Will validate whether the processing group, TransactionalExecutorProvider, and the converters required to construct statement factories/converters (the EventConverter and generic Converter) are set. If any of these are not provided an AxonConfigurationException is thrown.

Parameters:

builder - The JdbcSequencedDeadLetterQueue.Builder used to instantiate a JdbcSequencedDeadLetterQueue instance.

Method Details

builder

public static <E extends EventMessage>
JdbcSequencedDeadLetterQueue.Builder<E> builder()

Instantiate a builder to construct a JdbcSequencedDeadLetterQueue.

The following defaults are set by the builder:

• The table's schema defaults to a DeadLetterSchema.defaultSchema().
• The maximum amount of sequences defaults to 1024.
• The maximum sequence size defaults to 1024.
• The page size defaults to 100.
• The claim duration defaults to 30 seconds.

The processing group, TransactionalExecutorProvider, EventConverter, and generic Converter are hard requirements and should be provided.

Type Parameters:

E - The type of EventMessage maintained in the dead letter of this SequencedDeadLetterQueue.

Returns:

A Builder that can construct an JdbcSequencedDeadLetterQueue.

createSchema

public CompletableFuture<Void> createSchema(DeadLetterTableFactory tableFactory,
 @Nullable ProcessingContext context)

Performs the DDL queries to create the schema necessary for this SequencedDeadLetterQueue implementation.

Parameters:

tableFactory - The factory constructing the PreparedStatement to construct a DeadLetter entry table based on the configured DeadLetterSchema.

context - The ProcessingContext in which the schema will be created. May be null; when provided, the connection executor is obtained from the context.

Returns:

A CompletableFuture that completes when the schema has been created. Completes exceptionally with a JdbcException if the table or indices could not be created.

enqueue

public CompletableFuture<Void> enqueue(Object sequenceIdentifier,
 DeadLetter<? extends E> 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<E extends EventMessage>

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.

evict

public @NonNull CompletableFuture<Void> evict(@NonNull DeadLetter<? extends E> 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<E extends EventMessage>

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 E> letter,
 UnaryOperator<DeadLetter<? extends E>> 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<E extends EventMessage>

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<E extends EventMessage>

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 E>>> 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<E extends EventMessage>

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 @NonNull CompletableFuture<Iterable<Iterable<DeadLetter<? extends E>>>> 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<E extends EventMessage>

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<E extends EventMessage>

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 @NonNull 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<E extends EventMessage>

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<E extends EventMessage>

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 @NonNull 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<E extends EventMessage>

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(Function<DeadLetter<? extends E>,CompletableFuture<EnqueueDecision<E>>> processingTask,
 @Nullable ProcessingContext context)

Process a single sequence of enqueued dead letters with the given processingTask. 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!

Uses the EnqueueDecision returned by the processingTask to decide whether to SequencedDeadLetterQueue.evict(DeadLetter, ProcessingContext) or SequencedDeadLetterQueue.requeue(DeadLetter, UnaryOperator, ProcessingContext) the dead letter. 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.

Optimized override that retrieves available sequences one at a time (page size of 1) instead of in batches. Since no sequenceFilter is applied, every available sequence is a valid candidate, so there is no need to prefetch multiple sequences. This avoids unnecessary deserialization of dead letter entries that will never be evaluated.

Specified by:

process in interface SequencedDeadLetterQueue<E extends EventMessage>

Parameters:

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.

See Also:

• process(Predicate, Function, ProcessingContext)

process

public @NonNull CompletableFuture<Boolean> process(@NonNull Predicate<DeadLetter<? extends E>> sequenceFilter,
 @NonNull Function<DeadLetter<? extends E>,CompletableFuture<EnqueueDecision<E>>> 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<E extends EventMessage>

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 @NonNull CompletableFuture<Void> clear(@Nullable ProcessingContext context)

Description copied from interface: SequencedDeadLetterQueue

Clears out all dead letters present in this queue.

Specified by:

clear in interface SequencedDeadLetterQueue<E extends EventMessage>

Parameters:

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

Returns:

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