Transaction Outbox for Java

The Transaction Outbox pattern solves the problem of inconsistent state when a service needs to update a database and notify other services (e.g., via a message queue) in a single atomic operation.

In a standard @Transactional method, the database update and the message queue call are not part of the same transaction. If the message queue call succeeds but the database transaction fails, the system becomes inconsistent.

transaction-outbox provides a way to 'buffer' these external calls within your local database transaction. When you use the TransactionOutbox.schedule(Class<T> serviceClass) method, it returns a proxy. Any method calls made on this proxy are serialized and written to a database table (defaulting to TXNO_OUTBOX) as part of your existing database transaction.

Lifecycle of an outbox event:

1. Transaction Phase: The proxy call is recorded in the TXNO_OUTBOX table within the same transaction as your business logic. If the transaction rolls back, the outbox entry is also rolled back.
2. Immediate Execution: Once the transaction successfully commits, a thread attempts to make the real call to the target service asynchronously.
3. Retry/Mop-up: If the call fails or the application crashes, a background worker re-attempts the call a configurable number of times.
4. Dead Letter Queue: If all retries fail, the request is blocked (similar to a dead letter queue) for manual investigation and can be unblocked once the underlying issue is resolved.

Note: For this to ensure eventual consistency, both your local database updates and your downstream event handlers must be idempotent, as messages may be delivered multiple times during retries.

@PUT
@Path("/sales/{id}")
@Transactional
public void createWidget(@PathParam("id") SaleId saleId, Sale sale) {
  saleRepository.save(saleId, sale);
  // This proxy captures the call and writes it to the outbox table in the current transaction
  MessageQueue proxy = transactionOutbox.schedule(MessageQueue.class);
  proxy.postMessage(StockReductionEvent.of(saleId, sale.item(), sale.amount()));
  proxy.postMessage(IncomeEvent.of(saleId, sale.value()));
}

By default, tasks are processed in parallel, which may lead to out-of-order execution if tasks fail and are retried. To guarantee that tasks are processed in a specific sequence (e.g., for Kafka topics, Kinesis, or data replication), you can group them into a named "topic".

Within a topic, tasks are processed sequentially: a task must succeed before the next one in that topic begins. Note that different topics can still run in parallel.

Important Considerations:

• Latency: Tasks are processed via background flushing, so there is a delay between the transaction commit and task execution.
• Blocking: If a task fails, the entire topic is frozen until the task succeeds via retry. Use a TransactionOutboxListener to monitor failures.
• Throughput: A single topic is processed in a single-threaded fashion. To scale, use multiple topics.
• Kafka Optimization: For maximum throughput, use a topic name that combines the Kafka topic and partition name.

outbox.with().ordered("topic1").schedule(Service.class).process("red");
outbox.with().ordered("topic2").schedule(Service.class).process("green");
outbox.with().ordered("topic1").schedule(Service.class).process("blue");
outbox.with().ordered("topic2").schedule(Service.class).process("yellow");

Instead of injecting the TransactionOutbox directly into your business logic, you can create a pattern where specific services are injected with a @Remote binding annotation. This hides the outbox mechanics from the consumer.

1. Define a @BindingAnnotation (e.g., @Remote).
2. In your Guice module, bind the service using the @Remote annotation by calling outbox.schedule(Service.class).
3. Inject the service using the @Remote annotation.

// 1. Define the annotation
@Target({ElementType.PARAMETER, ElementType.METHOD, ElementType.FIELD})
@Retention(RetentionPolicy.RUNTIME)
@BindingAnnotation
public @interface Remote {}

// 2. Configure bindings
@Provides
@Remote
@Singleton
MyService remote(TransactionOutbox outbox) {
  return outbox.schedule(MyService.class);
}

@Provides
MyService local() {
  return new MyService();
}

// 3. Usage
@Inject
@Remote
private MyService myService;

void doSomething() {
  myService.doAThing(1, 2, 3);
}

To use the Spring extension for transaction-outbox, add the dependency to your build configuration.

Stable Releases

Use Maven Central for stable versions.

Development Snapshots

To use bleeding-edge releases, add the GitHub Package Repository to your pom.xml and configure authentication in your settings.xml using a GitHub Personal Access Token with read:package permissions.

<!-- Maven Stable -->
<dependency>
  <groupId>com.gruelbox</groupId>
  <artifactId>transactionoutbox-spring</artifactId>
  <version>7.1.750</version>
</dependency>

<!-- Gradle Stable -->
implementation 'com.gruelbox:transactionoutbox-spring:7.1.750'

<!-- Maven Snapshot Repository Configuration -->
<repositories>
  <repository>
    <id>github-transaction-outbox</id>
    <name>Gruelbox Github Repository</name>
    <url>https://maven.pkg.github.com/gruelbox/transaction-outbox</url>
  </repository>
</repositories>

An application requires a single, shared instance of TransactionOutbox. You configure it using a builder.

If you do not have an existing transaction management or dependency injection system, you can initialize a TransactionManager using connection details and create the outbox with a specific Persistor dialect.

To avoid requiring a default constructor on your target classes (e.g., MyClass), you can provide a custom Instantiator that knows how to create instances of your classes.

// Using connection details for an in-memory H2 database
TransactionManager transactionManager = TransactionManager.fromConnectionDetails(
    "org.h2.Driver", "jdbc:h2:mem:test;MV_STORE=TRUE", "test", "test"));

// Create the outbox
TransactionOutbox outbox = TransactionOutbox.builder()
  .transactionManager(transactionManager)
  .persistor(Persistor.forDialect(Dialect.H2))
  .build();

// Start a transaction and schedule work
transactionManager.inTransaction(tx -> {
  tx.connection().createStatement().execute("INSERT INTO...");
  outbox.schedule(MyClass.class).myMethod("Foo", "Bar");
});

Do not use Mockito to stub TransactionOutbox directly, as its contract is too complex. Instead, use the provided stub implementations for the builder arguments to create a lightweight, in-memory version of the outbox for testing.

Key stubs available:

• StubThreadLocalTransactionManager: A simple transaction manager for tests.
• StubPersistor: A persistor that does not save anything to a database.
• Submitter.withExecutor(...): Allows executing work immediately (e.g., using MoreExecutors.directExecutor()) rather than in a background thread.

// GIVEN
SomeService mockService = Mockito.mock(SomeService.class);

// Use stubs instead of mocks for the complex components
TransactionManager transactionManager = new StubThreadLocalTransactionManager();

TransactionOutbox outbox = TransactionOutbox.builder()
    .instantiator(Instantiator.using(clazz -> mockService)) // Return our mock
    .persistor(StubPersistor.builder().build()) // Doesn't save anything
    .submitter(Submitter.withExecutor(MoreExecutors.directExecutor())) // Execute all work in-line
    .clockProvider(() ->
        Clock.fixed(LocalDateTime.of(2020, 3, 1, 12, 0)
            .toInstant(ZoneOffset.UTC), ZoneOffset.UTC))
    .transactionManager(transactionManager)
    .build();

// WHEN
transactionManager.inTransaction(tx ->
   outbox.schedule(SomeService.class).doAThing(1));

// THEN
Mockito.verify(mockService).doAThing(1);

The default DefaultInvocationSerializer uses GSON and is intentionally restrictive to prevent deserialization vulnerabilities. It does not support polymorphic collections (e.g., List.of(1, "2", 3L)).

If you require full flexibility and trust your serialized data, use the transaction-outbox-jackson library. This allows you to use a Jackson ObjectMapper configured to handle the polymorphic nature of TransactionOutboxEntry.

// Add support for TransactionOutboxEntry to your normal application ObjectMapper
yourNormalSharedObjectMapper.registerModule(new TransactionOutboxJacksonModule());

// (Optional) support deep polymorphic requests
ObjectMapper objectMapper = yourNormalSharedObjectMapper.copy();
objectMapper.setDefaultTyping(TransactionOutboxJacksonModule.typeResolver());

// Serialize
String message = objectMapper.writeValueAsString(entry);

// Deserialize
TransactionOutboxEntry entry = objectMapper.readValue(message, TransactionOutboxEntry.class);