Nel mio lavoro uso da anni il cloud AWS per il rilascio di microservizi. Su più microservizi ho dovuto affrontare il problema di eseguire un'operazione a DB (su DynamoDB) e inviare un messaggio sulla coda SQS in modo atomico.
In questo articolo vedremo una breve introduzione al Transactional Outbox Pattern, che permette di risolvere proprio questo problema, e vedremo come implementarlo in un'app Spring Boot 4 che utilizza DynamoDB e SQS.
Utilizzeremo Floci per replicare un ambiente cloud AWS in locale. Trovi il repository del tutorial a questo link: https://github.com/vincenzo-racca/floci-aws-local.
Vedremo sia il pattern puro, che utilizza una tabella di outbox, sia la sua variante semplificata che usa i CDC (Change Data Capture) di DynamoDB Streams.
Il problema
Quando dobbiamo effettuare due o più operazioni su DB, e vogliamo che siano atomiche, dobbiamo utilizzare una transazione.
Facendo così, se una delle operazioni fallisce, tutte le altre vengono annullate (rollback).
Con Spring e JPA, possiamo utilizzare l'annotazione @Transactional per ottenere questo comportamento:
@Transactional
public void doSomething(...) {
personRepository.save(person);
orderRepository.save(order);
paymentRepository.save(payment);
}
Se uno dei salvataggi fallisce, viene fatto il rollback di tutte le operazioni precedenti.
Su DynamoDB, possiamo usare il metodo transactWriteItems per ottenere lo stesso comportamento.
Ma cosa succede se vogliamo eseguire una o più operazioni a DB e inviare subito dopo un messaggio su coda SQS? Le due operazioni non possono essere atomicamente legate, perché DynamoDB e SQS non supportano il 2PC (Two-Phase Commit).
- Se l'invio su coda viene eseguito dopo l'operazione a DB e l'invio fallisce, l'evento di pagamento non viene inviato.
- Se l'invio su coda viene eseguito prima dell'operazione a DB e l'operazione a DB fallisce, abbiamo inviato un evento di pagamento per un ordine che non esiste.
Transactional Outbox Pattern (versione con tabella di outbox)
Il Transactional Outbox Pattern risolve questo problema, introducendo due componenti:
- Una tabella di outbox, che contiene gli eventi da inviare su coda.
- Un componente che legge periodicamente la tabella di outbox e invia gli eventi su coda.
Insieme all'operazione di scrittura a DB dell'entità principale (in questo caso l'ordine), viene scritta anche l'entità di outbox. Le due operazioni a DB vengono eseguite in modo atomico, quindi se una delle due fallisce, entrambe vengono annullate.
Il processo di relay legge periodicamente la tabella di outbox, ne modifica lo stato (ad esempio da READY a SENT) e invia gli eventi su coda SQS.
Naturalmente, l'operazione di aggiornamento dello stato può fallire. In questo caso il processo di relay riproverà a inviare l'evento su coda SQS.
Proprio per la natura dell'at-least-once delivery di questo pattern, è importante che il consumer dell'evento
sia idempotente, cioè che nel caso riceva più volte lo stesso evento, non generi effetti collaterali.
Implementazione della versione con tabella di outbox con Spring Boot 4
La tabella outbox può essere implementata in modo generico cosicché possa essere riutilizzata da più microservizi:
@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;
}
}
Analizziamo l'entità OutboxEvent:
- eventId: è l'identificativo univoco dell'evento, generato come UUID.
- status: indica lo stato dell'evento (READY, SENT, DEAD).
- aggregateType: indica il tipo di entità a cui l'evento si riferisce (ad esempio Order).
- aggregateId: è l'identificativo dell'entità a cui l'evento si riferisce (ad esempio orderId).
- eventType: indica il tipo di evento (ad esempio OrderCreatedEvent).
- createdAt: indica la data di creazione dell'evento.
- nextAttemptAt: indica la data per cui il componente relay deve prendere in considerazione il record.
- aggregationTypeStatus: è una combinazione di aggregateType e status, utilizzata come chiave secondaria per la query degli eventi READY da inviare su coda SQS.
- attempts: indica il numero di tentativi di invio dell'evento su coda SQS, prima che il record sia considerato DEAD.
- payload: contiene il payload dell'evento, serializzato in JSON.
La partition key della tabella è l'eventId, mentre un secondo indice (GSI) ha aggregationTypeStatus come partition key e nextAttemptAt come sort key. Il GSI permette di eseguire la query degli eventi READY da inviare su coda SQS, ordinati per nextAttemptAt.
La valorizzazione del campo aggregationTypeStatus è fatta da una classe apposita:
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;
}
}
Come puoi vedere, il campo ha il formato aggregateType#status#shard, dove shard è un numero da 0 a SHARD_COUNT-1, calcolato come hash dell'eventId (quindi è sempre deterministico).
Questo permette di avere due vantaggi:
- Distribuendo gli eventi READY su più shard, si evita di avere un indice con un numero elevato di eventi READY (hot partition).
- Partizioni diverse possono essere processate in parallelo dal componente relay.
A questo punto, il componente Order Service deve salvare l'entità Order e l'entità OutboxEvent in un'unica transazione:
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);
});
}
Il componente relay può essere implementato con un metodo schedulato che effettua periodicamente una query paginata sull'indice:
@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));
}
Per ogni record recuperato, viene effettuato l'invio su coda SQS e successivamente viene aggiornato lo stato dell'evento in SENT.
Se l'invio su coda SQS fallisce, il record rimane in READY e il componente relay riproverà a inviarlo al prossimo ciclo schedulato.
Inoltre, siccome le query per GSI sono eventually consistent, il componente relay può effettuare l'update di stato condizionandolo al fatto che il record sia ancora in READY.
Infine, il record di tipo DEAD può essere gestito in modo più semplice, valorizzando la partizione sempre a zero, poiché non rappresenta una hot partition come avviene invece per i record con status READY.
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 (versione con CDC)
DynamoDB permette di abilitare gli stream su una tabella, così da poter ricevere gli eventi di CDC (Change Data Capture) sulla tabella:
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"
Possiamo poi utilizzare una Pipe EventBridge per ricevere gli eventi di tipo INSERT sullo 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"
In questo modo il codice si limita a salvare l'entità Order su DynamoDB. Il messaggio di pagamento viene inviato automaticamente sulla coda SQS tramite la Pipe EventBridge:
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);
}
Eventuali errori sulla sendMessage vengono gestiti con i retry della Pipe EventBridge, e se il messaggio non viene consegnato dopo un certo numero di tentativi, viene inviato sulla DLQ (Dead Letter Queue) configurata sulla Pipe.
Di seguito il diagramma che mostra il flusso di eventi della versione con CDC:
Approccio con tabella outbox vs approccio con CDC
Il vantaggio principale dell'approccio con CDC consiste nello spostare la responsabilità di leggere e inoltrare gli eventi dal codice a un’infrastruttura gestita.
Lo svantaggio principale dell'approccio con CDC consiste nel fatto che si è vincolati dal tipo di evento che viene generato e dalla disponibilità di quest'ultimo (per DynamoDB Streams l'evento è disponibile per 24 ore).
L'approccio con la tabella outbox permette di avere una maggiore flessibilità sul tipo di evento da inviare (decidi tu che tipo di evento persistere) a discapito di una gestione molto più complessa del codice.
Conclusioni
In questo tutorial abbiamo visto cos'è il Transactional Outbox Pattern e come implementarlo in un microservizio Spring Boot 4 che utilizza DynamoDB e SQS.
Abbiamo visto sia la versione con tabella di outbox, che permette di avere un maggiore controllo sul tipo di evento da inviare,
sia la versione con CDC, che permette di spostare la responsabilità di leggere e inoltrare gli eventi dal codice a un’infrastruttura gestita.
Possiamo dire che l'approccio con CDC è la scelta migliore quando ci basta propagare le modifiche ai dati con poca manutenzione e alta scalabilità. L'outbox è preferibile quando ti servono eventi di dominio ben progettati, controllo sul ciclo di vita e visibilità sullo stato di ogni messaggio, a costo di gestire più codice e più infrastruttura applicativa.
...