Transactional Outbox Pattern with Spring Boot 4, DynamoDB, and SQS

I've been using the AWS cloud to deploy microservices at work for years. In several microservices, I've had to tackle the problem of performing a database operation (on DynamoDB) and sending a message to an SQS queue atomically.

In this article, we'll briefly introduce the Transactional Outbox Pattern, which solves this very problem, and see how to implement it in a Spring Boot 4 application that uses DynamoDB and SQS.

We'll use Floci to replicate an AWS cloud environment locally. You can find the tutorial repository here: https://github.com/vincenzo-racca/floci-aws-local.

We'll cover both the original pattern, which uses an outbox table, and its simplified variant, which uses CDC (Change Data Capture) through DynamoDB Streams.

The problem

When we need to perform two or more database operations atomically, we must use a transaction.
This way, if one operation fails, all the others are rolled back.

With Spring and JPA, we can use the @Transactional annotation to achieve this behavior:

@Transactional
public void doSomething(...) {
    personRepository.save(person);
    orderRepository.save(order);
    paymentRepository.save(payment);
}

If one of the save operations fails, all previous operations are rolled back.
With DynamoDB, we can use the transactWriteItems method to achieve the same behavior.

But what happens if we want to perform one or more database operations and then immediately send a message to an SQS queue? The two operations cannot be tied together atomically because DynamoDB and SQS do not support 2PC (Two-Phase Commit).

The example above shows an order being saved to the database and a payment event being sent to an SQS queue:

• If the message is sent after the database operation and sending fails, the payment event is never sent.
• If the message is sent before the database operation and the database operation fails, we've sent a payment event for an order that doesn't exist.

Transactional Outbox Pattern (outbox table version)

The Transactional Outbox Pattern solves this problem by introducing two components:

• An outbox table containing the events to send to the queue.
• A component that periodically reads the outbox table and sends the events to the queue.

Along with writing the main entity (in this case, the order) to the database, we also write an outbox entity. The two database operations are performed atomically, so if either fails, both are rolled back.

The relay process periodically reads the outbox table, updates the event status (for example, from READY to SENT), and sends the events to an SQS queue.
Of course, the status update can fail. In that case, the relay process will try sending the event to the SQS queue again.
Because of this pattern's at-least-once delivery semantics, the event consumer must be idempotent: receiving the same event more than once must not produce additional side effects.

Implementing the outbox table version with Spring Boot 4

The outbox table can be implemented generically so that it can be reused across multiple microservices:

@DynamoDbBean
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class OutboxEvent {

    public static final String ORDER_AGGREGATION_TYPE = "Order";
    public static final long NEXT_ATTEMPT_DELAY_SECOND = 60; // 1 minute

    private String eventId;
    private OutboxEventStatus status;
    private String aggregateType;
    private String aggregateId;
    private String eventType;
    private Instant createdAt;
    private Long nextAttemptAt;
    private String aggregationTypeStatus;
    private int attempts;
    private String payload;
    private Long ttl;

    public enum OutboxEventStatus {
        READY, SENT, DEAD
    }

    @DynamoDbPartitionKey
    public String getEventId() {
        return eventId;
    }


    @DynamoDbSecondaryPartitionKey(indexNames = "aggregationTypeStatus-index")
    public String getAggregationTypeStatus() {
        return aggregationTypeStatus;
    }

    @DynamoDbSecondarySortKey(indexNames = "aggregationTypeStatus-index")
    public Long getNextAttemptAt() {
        return nextAttemptAt;
    }
}

Let's examine the OutboxEvent entity:

• eventId: the unique event identifier, generated as a UUID.
• status: the event status (READY, SENT, DEAD).
• aggregateType: the type of entity the event refers to (for example, Order).
• aggregateId: the identifier of the entity the event refers to (for example, orderId).
• eventType: the event type (for example, OrderCreatedEvent).
• createdAt: the event creation date.
• nextAttemptAt: the time at which the relay component should consider processing the record.
• aggregationTypeStatus: a combination of aggregateType and status, used as a secondary key to query READY events to send to the SQS queue.
• attempts: the number of attempts to send the event to the SQS queue before the record is considered DEAD.
• payload: the event payload, serialized as JSON.

The table's partition key is eventId, while a secondary index (GSI) uses aggregationTypeStatus as its partition key and nextAttemptAt as its sort key. The GSI lets us query READY events to send to the SQS queue, sorted by nextAttemptAt.

A dedicated class populates the aggregationTypeStatus field:

public final class OutboxKeys {

    public static final int SHARD_COUNT = 4;

    private OutboxKeys() {}

    public static String aggregationTypeStatus(String aggregateType,
                                               OutboxEventStatus status,
                                               String eventId) {
        int shard = Math.floorMod(eventId.hashCode(), SHARD_COUNT);
        return aggregateType + "#" + status + "#" + shard;
    }

    public static String aggregationTypeStatus(String aggregateType,
                                               OutboxEventStatus status,
                                               int shard) {
        return aggregateType + "#" + status + "#" + shard;
    }
}

As you can see, the field follows the format aggregateType#status#shard, where shard is a number from 0 to SHARD_COUNT-1, calculated from the hash of eventId (so it is always deterministic). This provides two advantages:

• Distributing READY events across multiple shards avoids concentrating a large number of READY events in a single index partition (a hot partition).
• Different partitions can be processed in parallel by the relay component.

At this point, the Order Service component must save the Order entity and the OutboxEvent entity in a single transaction:

private final DynamoDbTable<Order> orderTable;
private final DynamoDbTable<OutboxEvent> outboxEventTable;
private final DynamoDbEnhancedClient enhancedClient;
private final SqsTemplate sqsTemplate;
private final ObjectMapper objectMapper;

// ...
public void initialize() {
    Order order = Order.builder()
            .orderId(UUID.randomUUID().toString())
            .bookIsbn("1234567890")
            .price(BigDecimal.valueOf(19.99))
            .status(PENDING)
            .build();

    String eventId = UUID.randomUUID().toString();
    OutboxEvent outboxEvent = OutboxEvent.builder()
            .payload(objectMapper.writeValueAsString(order))
            .aggregateType(orderTable.tableName())
            .aggregateId(order.getOrderId())
            .eventId(eventId)
            .eventType("OrderCreated")
            .status(READY)
            .createdAt(Instant.now())
            .nextAttemptAt(Instant.now().getEpochSecond())
            .aggregationTypeStatus(aggregationTypeStatus(ORDER_AGGREGATION_TYPE,
                    READY, eventId))
            .build();

    log.info("Inserting order into DynamoDB: {}", order);

    enhancedClient.transactWriteItems(r -> {
        r.addPutItem(orderTable, order);
        r.addPutItem(outboxEventTable, outboxEvent);
    });
}

The relay component can be implemented using a scheduled method that periodically runs a paginated query on the index:

@Scheduled(fixedDelay = 5000)
public void processOutbox() {
    log.debug("Processing outbox events");
    DynamoDbIndex<OutboxEvent> index = outboxEventTable
            .index("aggregationTypeStatus-index");

    IntStream.range(0, SHARD_COUNT)
            .forEach(shard ->processPage(index, shard));
}

private void processPage(DynamoDbIndex<OutboxEvent> index, int shard) {
    String pk = aggregationTypeStatus(ORDER_AGGREGATION_TYPE, READY, shard);
    long now = Instant.now().getEpochSecond();

    QueryEnhancedRequest request = QueryEnhancedRequest.builder()
            .queryConditional(sortLessThanOrEqualTo(
                    k -> k.partitionValue(pk).sortValue(now)))
            .limit(100)
            .build();

    index.query(request)
            .forEach(page -> page.items().forEach(this::processEvent));
}

For each retrieved record, the event is sent to the SQS queue and its status is then updated to SENT.
If sending to the SQS queue fails, the record remains READY, and the relay component will try again on the next scheduled run.
Also, since GSI queries are eventually consistent, the relay component can make the status update conditional on the record still being READY.

Finally, DEAD records can be handled more simply by always setting the partition to zero, since they do not create a hot partition in the same way as records with READY status.

private void processEvent(OutboxEvent outboxEvent) {

    try {
        log.info("Processing outbox event: {}", outboxEvent);
        Order order = objectMapper.readValue(outboxEvent.getPayload(), Order.class);
        PaymentEvent paymentEvent = new PaymentEvent();
        paymentEvent.setOrderId(order.getOrderId());
        paymentEvent.setAmount(order.getPrice());
        sqsTemplate.send("from-order-to-payment-queue", paymentEvent);

        outboxEvent.setStatus(SENT);
        outboxEvent.setAggregationTypeStatus(aggregationTypeStatus(
                ORDER_AGGREGATION_TYPE,
                SENT,
                outboxEvent.getEventId()));
        outboxEvent.setTtl(Instant.now().getEpochSecond() + 3600 * 24); // Set TTL to 1 day from now
    } catch (RuntimeException e) {
        log.warn("Failed to process outbox event {}", outboxEvent.getEventId(), e);
        outboxEvent.setAttempts(outboxEvent.getAttempts() + 1);
        outboxEvent.setNextAttemptAt(Instant.now()
                .plusSeconds(NEXT_ATTEMPT_DELAY_SECOND)
                .getEpochSecond());
        if (outboxEvent.getAttempts() >= 3) {
            outboxEvent.setStatus(DEAD);
            outboxEvent.setAggregationTypeStatus(aggregationTypeStatus(
                    ORDER_AGGREGATION_TYPE,
                    DEAD,
                    0));
            log.error("Outbox event {} has failed 3 times and is marked as DEAD", outboxEvent.getEventId());
        }
    }
    finally {
        doConditionalUpdate(outboxEvent);
    }
}

private void doConditionalUpdate(OutboxEvent outboxEvent) {
    try {
        outboxEventTable.updateItem(r -> r.item(outboxEvent)
                .conditionExpression(Expression.builder()
                        .expression("#s = :ready")
                        .putExpressionName("#s", "status")
                        .putExpressionValue(":ready", AttributeValue.fromS("READY"))
                        .build()));
    }
    catch (ConditionalCheckFailedException e){
        log.debug("Outbox event {} already handled by another instance", outboxEvent.getEventId());
    }

}

Transactional Outbox Pattern (CDC version)

DynamoDB lets us enable streams on a table so that we can receive CDC (Change Data Capture) events for that table:

aws dynamodb create-table \
    --table-name order \
    --attribute-definitions AttributeName=orderId,AttributeType=S \
    --key-schema AttributeName=orderId,KeyType=HASH \
    --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 \
    --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
    --endpoint-url "$endpoint_url"

We can then use an EventBridge Pipe to receive INSERT events from the stream:

stream_arn=$(aws dynamodb describe-table \
    --table-name order \
    --query 'Table.LatestStreamArn' --output text \
    --endpoint-url "$endpoint_url")
    
echo "Create DynamoDB stream destination queue"
queue_url=$(aws sqs create-queue \
    --queue-name from-order-to-payment-queue \
    --query QueueUrl --output text \
    --endpoint-url "$endpoint_url")
queue_arn=$(aws sqs get-queue-attributes \
    --queue-url "$queue_url" \
    --attribute-names QueueArn \
    --query 'Attributes.QueueArn' --output text \
    --endpoint-url "$endpoint_url")
    
echo "Create EventBridge Pipe from DynamoDB INSERT events to SQS"
aws pipes create-pipe \
    --name order-inserts-to-sqs \
    --desired-state RUNNING \
    --source "$stream_arn" \
    --source-parameters \
        '{
            "DynamoDBStreamParameters": {
                "StartingPosition": "TRIM_HORIZON",
                "BatchSize": 1
            },
            "FilterCriteria": {
                "Filters": [
                    {
                        "Pattern": "{\"eventName\":[\"INSERT\"]}"
                    }
                ]
            }
        }' \
    --target "$queue_arn" \
    --target-parameters \
        '{
            "InputTemplate": "{\"orderId\":<$.dynamodb.NewImage.orderId.S>,\"amount\":<$.dynamodb.NewImage.price.N>}"
        }' \
    --role-arn arn:aws:iam::000000000000:role/order-inserts-to-sqs \
    --endpoint-url "$endpoint_url"

This way, the code only needs to save the Order entity to DynamoDB. The payment message is automatically sent to the SQS queue through the EventBridge Pipe:

public void initialize() {
  Order order = Order.builder()
          .orderId(UUID.randomUUID().toString())
          .bookIsbn("1234567890")
          .price(BigDecimal.valueOf(19.99))
          .status(PENDING)
          .build();

  PutItemEnhancedRequest<Order> request = PutItemEnhancedRequest
          .builder(Order.class)
          .item(order)
          .build();
  log.info("Inserting order into DynamoDB: {}", order);
  orderTable.putItem(request);
}

Any sendMessage errors are handled by the EventBridge Pipe's retry mechanism. If the message is not delivered after a certain number of attempts, it is sent to the DLQ (Dead Letter Queue) configured on the Pipe.

The following diagram shows the event flow for the CDC version:

Outbox table approach vs. CDC approach

The main advantage of the CDC approach is that it shifts the responsibility for reading and forwarding events from application code to managed infrastructure.

The main disadvantage of the CDC approach is that you are constrained by the type of event generated and how long it remains available (DynamoDB Streams events are available for 24 hours).

The outbox table approach provides more flexibility over the type of event to send (you decide which event to persist), at the cost of much more complex code to manage.

Conclusions

In this tutorial, we explored the Transactional Outbox Pattern and how to implement it in a Spring Boot 4 microservice that uses DynamoDB and SQS.
We covered both the outbox table version, which gives us more control over the type of event to send, and the CDC version, which shifts the responsibility for reading and forwarding events from application code to managed infrastructure.

The CDC approach is the better choice when all we need is to propagate data changes with little maintenance and high scalability. The outbox approach is preferable when you need well-designed domain events, lifecycle control, and visibility into each message's status, at the cost of managing more code and more application infrastructure.