gruelbox/transaction-outbox

Reliable eventual consistency for Microservices

transaction-outbox

A flexible implementation of the Transaction Outbox Pattern for Java. TransactionOutbox has a clean, extensible API, very few dependencies and plays nicely with a variety of database platforms, transaction management approaches and application frameworks. Every aspect is highly configurable or overridable. It features out-of-the-box support for Spring DI, Spring Txn, Guice, MySQL 5 & 8, PostgreSQL 11-16, Oracle 18 & 21, MS SQL Server 2017 and H2.

Contents

1. Why do I need it?
2. Installation
   1. Requirements
   2. Stable releases
   3. Development snapshots
3. Basic Configuration
   1. No existing transaction manager or dependency injection
   2. Spring
   3. Guice
   4. jOOQ
4. Set up the background worker
5. Managing the "dead letter queue"
6. Advanced
   1. Topics and FIFO ordering
   2. The nested outbox pattern
   3. Idempotency protection
   4. Delayed/scheduled processing
   5. Flexible serialization
   6. Clustering
7. Configuration reference
8. Stubbing in tests

Why do I need it?

This article explains the concept in an abstract manner, but let's say we have a microservice that handles point-of-sale and need to implement a REST endpoint to record a sale. We end up with this:

Attempt 1

@POST
@Path("/sales")
@Transactional
public SaleId createWidget(Sale sale) {
  var saleId = saleRepository.save(sale);
  messageQueue.postMessage(StockReductionEvent.of(sale.item(), sale.amount()));
  messageQueue.postMessage(IncomeEvent.of(sale.value()));
  return saleId;
}

The SaleRepository handles recording the sale in the customer's account, the StockReductionEvent goes off to our warehouse service, and the IncomeEvent goes to our financial records service (let's ignore the potential flaws in the domain modelling for now).

There's a big problem here: the @Transactional annotation is a lie (no, really). It only really wraps the SaleRepository call, but not the two event postings. This means that we could end up sending the two events and fail to actually commit the sale. Our system is now inconsistent.

Attempt 2 - Use Idempotency

We could make whole method idempotent and re-write it to work a bit more like this:

@PUT
@Path("/sales/{id}")
public void createWidget(@PathParam("id") SaleId saleId, Sale sale) {
  saleRepository.saveInNewTransaction(saleId, sale);
  messageQueue.postMessage(StockReductionEvent.of(saleId, sale.item(), sale.amount()));
  messageQueue.postMessage(IncomeEvent.of(saleId, sale.value()));
}

This is better. As long as the caller keeps calling the method until they get a success, we can keep re-saving and re-sending the messages without any risk of duplicating work. This works regardless of the order of the calls (and in any case, there may be good reasons of referential integrity to fix the order).

The problem is that they might stop trying, and if they do, we could end up with only part of this transaction completed. If this is a public API, we can't force clients to use it correctly.

We also still have another problem: external calls are inherently more vulnerable to downtime and performance degredation. We could find our service rendered unresponsive or failing if they are unavailable. Ideally, we would like to "buffer" these external calls within our service safely until our downstream dependencies are available.

Attempt 3 - Transaction Outbox

Idempotency is a good thing, so let's stick with the PUT. Here is the same example, using Transaction Outbox:

@PUT
@Path("/sales/{id}")
@Transactional
public void createWidget(@PathParam("id") SaleId saleId, Sale sale) {
  saleRepository.save(saleId, sale);
  MessageQueue proxy = transactionOutbox.schedule(MessageQueue.class);
  proxy.postMessage(StockReductionEvent.of(saleId, sale.item(), sale.amount()));
  proxy.postMessage(IncomeEvent.of(saleId, sale.value()));
}

Here's what happens:

• When you create an instance of TransactionOutbox (see Basic Configuration), it will, by default, automatically create two database tables, TXNO_OUTBOX and TXNO_VERSION, and then keep these synchronized with schema changes as new versions are released. Note: this is the default behaviour on a SQL database, but is completely overridable if you are using a different type of data store or don't want a third party library managing your database schema. See Configuration reference.
• TransactionOutbox creates a proxy of MessageQueue. Any method calls on the proxy are serialized and written to the TXNO_OUTBOX table (by default) in the same transaction as the SaleRepository call. The call returns immediately rather than actually invoking the real method.
• If the transaction rolls back, so do the serialized requests.
• Immediately after the transaction is successfully committed, another thread will attempt to make the real call to MessageQueue asynchronously.
• If that call fails, or the application dies before the call is attempted, a background "mop-up" thread will re-attempt the call a configurable number of times, with configurable time between each, before blocking the request and firing and event for it to be investigated (similar to a dead letter queue).
• Blocked requests can be easily unblocked again once the underlying issue is resolved.

Our service is now resilient and explicitly eventually consistent, as long as all three elements (SaleRepository and the downstream event handlers) are idempotent, since those messages will be attempted repeatedly until confirmed successful, which means they could occur multiple times.

If you find yourself wondering why bother with the queues now? You're quite right. As we now have outgoing buffers, we already have most of the benefits of middleware (at least for some use cases). We could replace the calls to a message queue with direct queues to the other services' load balancers and switch to a peer-to-peer architecture, if we so choose.

Note that for the above example to work, StockReductionEvent and IncomeEvent need to be included for serialization. See Configuration reference.

Installation

Requirements

• At least Java 11. Downgrading to requiring Java 8 is under consideration.
• Currently, MySQL, PostgreSQL, Oracle, MS SQL Server or H2 databases (pull requests to support other traditional RDMBS would be trivial. Beyond that, a SQL database is not strictly necessary for the pattern to work, merely a data store with the concept of a transaction spanning multiple mutation operations).
• Database access via JDBC (In principle, JDBC should not be required - alternatives such as R2DBC are under investigation - but the API is currently tied to it)
• Native transactions (not JTA or similar).
• (Optional) Proxying non-interfaces requires ByteBuddy and for proxying classes without default constructors Objenesis to be added as a dependency

Stable releases

The latest stable release is available from Maven Central. Stable releases are sort-of semantically versioned. That is, they follow semver in every respect other than that the version numbers are not monotically increasing. The project uses continuous delivery and selects individual stable releases to promote to Central, so Central releases will always be spaced apart numerically. The important thing, though, is that dependencies should be safe to upgrade as long as the major version number has not increased.

Maven

<dependency>
  <groupId>com.gruelbox</groupId>
  <artifactId>transactionoutbox-core</artifactId>
  <version>5.5.447</version>
</dependency>

Gradle

implementation 'com.gruelbox:transactionoutbox-core:5.5.447'

Development snapshots

Maven Central is updated regularly. However, if you want to stay at the bleeding edge, you can use continuously-delivered releases from Github Package Repository. These can be used from your production builds since they will never be deleted (unlike SNAPSHOTs).

Maven

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

You will need to authenticate with Github to use Github Package Repository. Create a personal access token in your GitHub settings. It only needs read:package permissions. Then add something like the following in your Maven settings.xml:

<servers>
    <server>
        <id>github-transaction-outbox</id>
        <username>${env.GITHUB_USERNAME}</username>
        <password>${env.GITHUB_TOKEN}</password>
    </server>
</servers>

The above example uses environment variables, allowing you to keep the credentials out of source control, but you can hard-code them if you know what you're doing.

Gradle

repositories {
    maven {
        name = "github-transaction-outbox"
        url = uri("https://maven.pkg.github.com/gruelbox/transaction-outboxY")
        credentials {
            username = $githubUserName
            password = $githubToken
        }
    }
}

Basic Configuration

An application needs a single, shared instance of TransactionOutbox, which is configured using a builder on construction. This takes some time to get right, particularly if you already have a transaction management solution in your application.

No existing transaction manager or dependency injection

If you have no existing transaction management, connection pooling or dependency injection, here's a quick way to get started:

// Use 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
transactionManager.inTransaction(tx -> {
  // Save some stuff
  tx.connection().createStatement().execute("INSERT INTO...");
  // Create an outbox request
  outbox.schedule(MyClass.class).myMethod("Foo", "Bar"));
});

Alternatively, you could create the TransactionManager from a DataSource, allowing you to use a connection pooling DataSource such as Hikari:

TransactionManager transactionManager = TransactionManager.fromDataSource(dataSource);

In this default configuration, MyClass must have a default constructor so the "real" implementation can be constructed at the point the method is actually invoked (which might be on another day on another instance of the application). However, you can avoid this requirement by providing an Instantiator on every instance of your application that knows how to create the objects:

TransactionOutbox outbox = TransactionOutbox.builder()
  .instantiator(Instantiator.using(clazz -> createInstanceOf(clazz)))
  .build();

Spring

See transaction-outbox-spring, which integrates Spring's DI and/or transaction management with TransactionOutbox.

Guice

See transaction-outbox-guice, which integrates Guice DI TransactionOutbox.

jOOQ

See transaction-outbox-jooq, which integrates jOOQ transaction management with TransactionOutbox.

Oracle

Oracle database compatibility requires to configure Oracle jdbc driver using following VM argument : -Doracle.jdbc.javaNetNio=false

Set up the background worker

At the moment, if any work fails first time, it won't be retried. All we need to add is a background thread that repeatedly calls TransactionOutbox.flush() to pick up and reprocess stale work.

How you do this is up to you; it very much depends on how background processing works in your application (a reactive solution will be very different to one based on Guava Service, for example). However, here is a simple example:

Thread backgroundThread = new Thread(() -> {
  while (!Thread.interrupted()) {
    try {
      // Keep flushing work until there's nothing left to flush
      while (outbox.flush()) {}
    } catch (Exception e) {
      log.error("Error flushing transaction outbox. Pausing", e);
    }
    try {
       // When we run out of work, pause for a minute before checking again
       Thread.sleep(60_000);
    } catch (InterruptedException e) {
       break;
    }
  }
});

// Startup
backgroundThread.start();

// Shut down
backgroundThread.interrupt();
backgroundThread.join();

flush() is designed to handle concurrent use on databases that support SKIP LOCKED, such as Postgres and MySQL 8+. Feel free to run this as often as you like (within reason, e.g. once a minute) on every instance of your application. This can have the benefit of spreading work across multiple instances when the work backlog is extremely high, but is not as effective as a proper clustering approach.

However, multiple concurrent calls to flush() can cause lock timeout errors on databases without SKIP LOCKED support, such as MySQL 5.7. This is harmless, but will cause a lot of log noise, so you may prefer to run on a single instance at a time to avoid this.

Managing the "dead letter queue"

Work might be retried too many times and enter a blocked state. You should set up an alert to allow you to manage this when it occurs, resolve the issue and unblock the work, since the work not being complete will usually be a sign that your system is out of sync in some way.

TransactionOutbox.builder()
    ...
    .listener(new TransactionOutboxListener() {
        @Override
        public void blocked(TransactionOutboxEntry entry, Throwable cause) {
           // Spring example
           applicationEventPublisher.publishEvent(new TransactionOutboxBlockedEvent(entry.getId(), cause);
        }
    })
    .build();

To mark the work for reprocessing, just use TransactionOutbox.unblock(). Its failure count will be marked back down to zero and it will get reprocessed on the next call to flush():

transactionOutboxEntry.unblock(entryId);

Or if using a TransactionManager that relies on explicit context (such as a non-thread local JooqTransactionManager):

transactionOutboxEntry.unblock(entryId, context);

A good approach here is to use the TransactionOutboxListener callback to post an interactive Slack message - this can operate as both the alert and the "button" allowing a support engineer to submit the work for reprocessing.

Advanced

Topics and FIFO ordering

For some applications, the order in which tasks are processed is important, such as when:

• using the outbox to write to a FIFO queue, Kafka or AWS Kinesis topic; or
• data replication, e.g. when feeding a data warehouse or distributed cache.

In these scenarios, the default behaviour is unsuitable. Tasks are usually processed in a highly parallel fashion. Even if the volume of tasks is low, if a task fails and is retried later, it can easily end up processing after some later task even if that later task was processed hours or even days after the failing one.

To avoid problems associated with tasks being processed out-of-order, you can order the processing of your tasks within a named "topic":

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");

No matter what happens:

• red will always need to be processed (successfully) before blue;
• green will always need to be processed (successfully) before yellow; but
• red and blue can run in any sequence with respect to green and yellow.

This functionality was specifically designed to allow outboxed writing to Kafka topics. For maximum throughput when writing to Kafka, it is advised that you form your outbox topic name by combining the Kafka topic and partition, since that is the boundary where ordering is required.

There are a number of things to consider before using this feature:

• Tasks are not processed immediately when submitting, as normal, and are processed by background flushing only. This means there will be an increased delay between the source transaction being committed and the task being processed, depending on how your application calls TransactionOutbox.flush().
• If a task fails, no further requests will be processed in that topic until a subsequent retry allows the failing task to succeed, to preserve ordered processing. This means it is possible for topics to become entirely frozen in the event that a task fails repeatedly. For this reason, it is essential to use a TransactionOutboxListener to watch for failing tasks and investigate quickly. Note that other topics will be unaffected.
• TransactionOutboxBuilder.blockAfterAttempts is ignored for all tasks that use this option.
• A single topic can only be processed in single-threaded fashion, but separate topics can be processed in parallel. If your tasks use a small number of topics, scalability will be affected since the degree of parallelism will be reduced.

The nested-outbox pattern

In practice it can be extremely hard to guarantee that an entire unit of work is idempotent and thus suitable for retry. For example, the request might be to "update a customer record" with a new address, but this might record the change to an audit history table with a fresh UUID, the current date and time and so on, which in turn triggers external changes outside the transaction. The parent customer update request may be idempotent, but the downstream effects may not be.

To tackle this, TransactionOutbox supports a use case where outbox requests spawn further outbox requests, along with a layer of additional idempotency protection for particularly diffcult cases. The nested pattern works as follows:

• Modify the customer record: outbox.schedule(CustomerService.class).update(newDetails)
• The update method spawns a new outbox request to process the downstream effect: outbox.schedule(AuditService.class).audit("CUSTOMER_UPDATED", UUID.randomUUID(), Instant.now(), newDetails.customerId())

Now, if any part of the top-level request throws, nothing occurs. If the top level request succeeds, an idempotent request to create the audit record will retry safely.

Idempotency protection

A common use case for TransactionOutbox is to receive an incoming request (such as a message from a message queue), acknowledge it immediately and process it asynchronously, for example:

public class FooEventHandler implements SQSEventHandler<ThingHappenedEvent> {

  @Inject private TransactionOutbox outbox;

  public void handle(ThingHappenedEvent event) {
    outbox.schedule(FooService.class).handleEvent(event.getThingId());
  }
}

However, incoming transports, whether they be message queues or APIs, usually need to rely on idempotency in message handlers (for the same reason that outgoing requests from outbox also rely on idempotency). This means the above code could get called twice.

As long as FooService.handleEvent() is idempotent itself, this is harmless, but we can't always assume this. The incoming message might be a broadcast, with no knowledge of the behaviour of handlers and therefore no way of pre-generating any new record ids the handler might need and passing them in the message.

To protect against this, TransactionOutbox can automatically detect duplicate requests and reject them with AlreadyScheduledException. Records of requests are retained up to a configurable threshold (see below).

To use this, use the call pattern:

outbox.with()
  .uniqueRequestId("context-clientid")
  .schedule(Service.class)
  .process("Foo");

Where context-clientid is a globally-unique identifier derived from the incoming request. Such ids are usually available from queue middleware as message ids, or if not you can require as part of the incoming API (possibly with a tenant prefix to ensure global uniqueness across tenants).

Delayed/scheduled processing

To delay execution of a task, use:

outbox.with()
  .delayForAtLeast(Duration.of(5, MINUTES))
  .schedule(Service.class)
  .process("Foo");

There are some caveats around how accurate timing is. See the JavaDoc on the delayForAtLeast method for more information.

This is particularly useful when combined with the nested outbox pattern for creating polling/repeated or recursive tasks to throttle prcessing.

Flexible serialization (beta)

Most people will use the default persistor, DefaultPersistor, to persist tasks to a relational database. This uses DefaultInvocationSerializer by default, which in turn uses GSON to serialize as JSON. DefaultInvocationSerializer is extremely limited by design, with a small list of allowed classes in method arguments. You can extend the list of support types by calling serializableTypes in its builder, but it will always be restricted to this global list. This is by design, to avoid building a deserialization of untrusted data vulnerability into the library.

Furthermore, there is no support for the case where run-time and compile-time types differ, such as in polymorphic collections. The following will always fail with DefaultInvocationSerializer:

outbox.schedule(Service.class).processList(List.of(1, "2", 3L));

However, if you completely trust your serialized data (for example, your developers don't have write access to your production database, and the access credentials are well guarded) then you may prefer to have 100% flexibility, with no need to declare the types used and the ability to use any combination of run-time and compile-time types.

See transaction-outbox-jackson, which uses a specially-configured Jackson ObjectMapper to achieve this.

Clustering

The default mechanism for running tasks (either immediately, or when they are picked up by background processing) is via a java.concurrent.Executor, which effectively does the following:

executor.execute(() -> outbox.processNow(transactionOutboxEntry));

This offloads processing to a background thread on the application instance on which the task was picked up. Under high load, this can mean thousands of tasks being picked up from the database queue and submitted at the same time on the same application instance, even if there are 20 instances of the application, effectively limiting the total rate of processing to what a single instance can handle.

If you want to instead push the work for processing by any of your application instances, thus spreading the work around a cluster, there are multiple approaches, just some of which are listed below:

• An HTTP endpoint on a load-balanced DNS with service discovery (such as a container orchestrator e.g. Kubernetes or Nomad)
• A shared queue (AWS SQS, ActiveMQ, ZeroMQ)
• A lower-level clustering/messaging toolkit such as JGroups.

All of these can be implemented as follows:

When defining the TransactionOutbox, replace ExecutorSubmitter with something which serializes a TransactionOutboxEntry and ships it to the remote queue/address. Here's what configuration might look for a RestApiSubmitter which ships the request to a load-balanced endpoint hosted on Nomad/Consul:

TransactionOutbox outbox = TransactionOutbox.builder().submitter(restApiSubmitter)

It is strongly advised that you use a local executor in-line, to ensure that if there are communications issues with your endpoint or queue, it doesn't fail the calling thread. Here is an example using Feign:

@Slf4j
class RestApiSubmitter implements Submitter {

  private final FeignResource feignResource;
  private final ExecutorService localExecutor;
  private final Provider<TransactionOutbox> outbox;

  @Inject
  RestApiExecutor(String endpointUrl, ExecutorService localExecutor, ObjectMapper objectMapper, Provider<TransactionOutbox> outbox) {
    this.feignResource = Feign.builder()
      .decoder(new JacksonDecoder(objectMapper))
      .target(GitHub.class, "https://api.github.com");;
    this.localExecutor = localExecutor;
    this.outbox = outbox;
  }

  @Override
  public void submit(TransactionOutboxEntry entry, Consumer<TransactionOutboxEntry> leIgnore) {
    try {
      localExecutor.execute(() -> processRemotely(entry));
      log.info("Queued {} to be sent for remote processing", entry.description());
    } catch (RejectedExecutionException e) {
      log.info("Will queue {} for processing when local executor is available", entry.description());
    } catch (Exception e) {
      log.warn("Failed to queue {} for execution at {}. It will be re-attempted later.", entry.description(), url, e);
    }
  }

  private void processRemotely(TransactionOutboxEntry entry) {
    try {
      feignResource.process(entry);
      log.info("Submitted {} for remote processing at {}", entry.description(), url);
    } catch (Exception e) {
      log.warn(
        "Failed to submit {} for remote processing at {}. It will be re-attempted later.",
        entry.description(),
        url,
        e
      );
    }
  }
  
  public interface FeignResource {
    @RequestLine("POST /outbox/process")
    void process(TransactionOutboxEntry entry);
  }
  
}

Then listen on your communication mechanism for incoming serialized TransactionOutboxEntrys, and push them to a normal local ExecutorSubmitter. Here's what a JAX-RS example might look like:

@POST
@Path("/outbox/process")
void processOutboxEntry(String entry) {
  TransactionOutboxEntry entry = somethingWhichCanSerializeTransactionOutboxEntries.deserialize(entry);
  Submitter submitter = ExecutorSubmitter.builder().executor(localExecutor).logLevelWorkQueueSaturation(Level.INFO).build();
  submitter.submit(entry, outbox.get()::processNow);
}

This whole approach is complicated a little by the inherent difficulty in serializing and deserializing a TransactionOutboxEntry, which is extremely polymorphic in nature. A reference approach is provided by transaction-outbox-jackson, which provides the features necessary to make a Jackson ObjectMapper able to handle the work. With that on the classpath you can use an ObjectMapper as follows:

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

// (Optional) support deep polymorphic requests - for this we need to copy the object
// mapper so it doesn't break the way the rest of your application works
ObjectMapper objectMapper = yourNormalSharedObjectMapper.copy();
objectMapper.setDefaultTyping(TransactionOutboxJacksonModule.typeResolver());

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

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

Armed with the above, happy clustering!

Configuration reference

This example shows a number of other configuration options in action:

TransactionManager transactionManager = TransactionManager.fromDataSource(dataSource);

TransactionOutbox outbox = TransactionOutbox.builder()
    // The most complex part to set up for most will be synchronizing with your existing transaction
    // management. Pre-rolled implementations are available for jOOQ and Spring (see above for more information)
    // and you can use those examples to synchronize with anything else by defining your own TransactionManager.
    // Or, if you have no formal transaction management at the moment, why not start, using transaction-outbox's
    // built-in one?
    .transactionManager(transactionManager)
    // Modify how requests are persisted to the database. For more complex modifications, you may wish to subclass
    // DefaultPersistor, or create a completely new Persistor implementation.
    .persistor(DefaultPersistor.builder()
        // Selecting the right SQL dialect ensures that features such as SKIP LOCKED are used correctly.
        .dialect(Dialect.POSTGRESQL_9)
        // Override the table name (defaults to "TXNO_OUTBOX")
        .tableName("transactionOutbox")
        // Shorten the time we will wait for write locks (defaults to 2)
        .writeLockTimeoutSeconds(1)
        // Disable automatic creation and migration of the outbox table, forcing the application to manage
        // migrations itself
        .migrate(false)
        // Allow the SaleType enum and Money class to be used in arguments (see example below)
        .serializer(DefaultInvocationSerializer.builder()
            .serializableTypes(Set.of(SaleType.class, Money.class))
            .build())
        .build())
    // GuiceInstantiator and SpringInstantiator are great if you are using Guice or Spring DI, but what if you
    // have your own service locator? Wire it in here. Fully-custom Instantiator implementations are easy to
    // implement.
    .instantiator(Instantiator.using(myServiceLocator::createInstance))
    // Change the log level used when work cannot be submitted to a saturated queue to INFO level (the default
    // is WARN, which you should probably consider a production incident). You can also change the Executor used
    // for submitting work to a shared thread pool used by the rest of your application. Fully-custom Submitter
    // implementations are also easy to implement, for example to cluster work.
    .submitter(ExecutorSubmitter.builder()
        .executor(ForkJoinPool.commonPool())
        .logLevelWorkQueueSaturation(Level.INFO)
        .build())
    // Lower the log level when a task fails temporarily from the default WARN.
    .logLevelTemporaryFailure(Level.INFO)
    // 10 attempts at a task before blocking it.
    .blockAfterAttempts(10)
    // When calling flush(), select 0.5m records at a time.
    .flushBatchSize(500_000)
    // Flush once every 15 minutes only
    .attemptFrequency(Duration.ofMinutes(15))
    // Include Slf4j's Mapped Diagnostic Context in tasks. This means that anything in the MDC when schedule()
    // is called will be recreated in the task when it runs. Very useful for tracking things like user ids and
    // request ids across invocations.
    .serializeMdc(true)
    // Sets how long we should keep records of requests with a unique request id so duplicate requests
    // can be rejected. Defaults to 7 days.
    .retentionThreshold(Duration.ofDays(1))
    // We can intercept task successes, single failures and blocked tasks. The most common use is to catch blocked tasks
    // and raise alerts for these to be investigated. A Slack interactive message is particularly effective here
    // since it can be wired up to call unblock() automatically.
    .listener(new TransactionOutboxListener() {

      @Override
      public void success(TransactionOutboxEntry entry) {
        eventPublisher.publish(new OutboxTaskProcessedEvent(entry.getId()));
      }

      @Override
      public void blocked(TransactionOutboxEntry entry, Throwable cause) {
        eventPublisher.publish(new BlockedOutboxTaskEvent(entry.getId()));
      }

    })
    .build();

// Usage example, using the in-built transaction manager
MDC.put("SESSIONKEY", "Foo");
try {
  transactionManager.inTransaction(tx -> {
    writeSomeChanges(tx.connection());
    outbox.schedule(getClass())
        .performRemoteCall(SaleType.SALE, Money.of(10, Currency.getInstance("USD")));
  });
} finally {
  MDC.clear();
}

Stubbing in tests

TransactionOutbox should not be directly stubbed (e.g. using Mockito); the contract is too complex to stub out.

Instead, stubs exist for the various arguments to the builder, allowing you to build a TransactionOutbox with minimal external dependencies which can be called and verified in tests.

// GIVEN

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

// Also see StubParameterContextTransactionManager
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)) // Fix the clock (not necessary here)
    .transactionManager(transactionManager)
    .build();

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

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

Depending on the type of test, you may wish to use a real Persistor such as DefaultPersistor (if there's a real database available) or a real, multi-threaded Submitter. That's up to you.

transaction-outbox's Issues

Performance Load Test Suite

Looking to add some performance test suites to make performance testing easier and to hopefully give users tools to identify the outbox load that could be supported by their database. Was thinking to implement this in JMeter but I would be open to other testing tools as well if there is a preference.

Objenesis and CGLIB could be optional dependences

Could work like Spring does and use them if they're on the classpath, otherwise refuse nicely to create proxies which need them.

Entry failure logging omits MDC context

Transaction Outbox helpfully supports serialising logging MDC and passing this to the thread that performs the processing. This can help immensely with tracing in environments that have a lot of concurrent transactions.

When a transaction outbox entry fails, it writes a log message along the lines of:

Temporarily failed to process entry e512a3e7-6c7c-4dd2-bfec-f11765deab51 : service.method("arg1", "arg2") [e512a3e7-6c7c-4dd2-bfec-f11765deab51]

Unfortunately, this logging is written without the MDC context for that transaction outbox entry. This can make it difficult to relate the error to the processing that preceded it, and diminishes the value of the MDC serialisation feature. This seems like a bug - it would surely be preferable for the exception to be logged with the invocation's MDC?

This issue appears to be because the MDC entry/exit happens in Invocation.invoke, but the error handling is outside that, in TransactionOutboxImpl.processNow. I think this could be fixed by pulling the MDC entry/exit logic up to wrap the entire contents of 'processNow'.

I may attempt a PR, but wanted to raise this nonetheless.

Integration Problems with Spring Boot Application - looking for more code samples

Hi folks,

I am a bit hesitant to create a report but I'm stuck with Spring's Boot configuration magic (version 3.1.6) - well, sort of expected :)

Are there any hints, more code samples, mailing lists or whatsoever?! I'm really not eager to write my own outbox code because it is hard when aiming at production quality.

With the Spring integration library I got the outbox working but my Spring Boot test are failing since Flyway decided to ignore my Postgres test container

Caused by: org.flywaydb.core.internal.exception.FlywaySqlException: Unable to obtain connection from database: Connection to localhost:5435 refused. Check that the hostname and port are correct and that the postmaster is accepting TCP/IP connections.
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
SQL State  : 08001
Error Code : 0
Message    : Connection to localhost:5435 refused. Check that the hostname and port are correct and that the postmaster is accepting TCP/IP connections.

When using the core library alone the Spring Boot test are working but the the library is unable to find the current transaction (using transaction manager and Hikari DS) - guess that stems from byte code magic and calling the wrong proxy.

SpringTransactionManager does not cache insert PreparedStatements

In SpringTransactionManager, any time Transaction.prepareBatchStatement is called, a new PreparedStatement is opened. In other implementations, the statements are cached in the transaction and re-used for the same SQL statement.

Harmless functionally, but affects performance.

Tests have produced a case where a message was processed out of order

https://github.com/gruelbox/transaction-outbox/actions/runs/8772003689/job/24075529532?pr=622#step:4:304

This might be a false alarm (e.g. caused by an in-memory race condition in the test) so investigating further.

This could be a serious issue though so raising as an issue for visibility.

Better SPI documentation

How to create:

• TransactionManagers
• Persistors
• Instantiators

transactionoutbox-spring depends on hibernate-core

I am using Spring, but not Hibernate, and the code in SpringTransactionManager won't work for me. There is no SessionFactory from which to fetch the active Connection.

I think the answer is just to inject the DataSource, and use connections from that. You should be able to require that this DataSource has been either
a) wrapped in TransactionAwareDataSourceProxy in a non-JTA context
b) is a JTA connection that will handle the transactionality for you.

If this make sense, I might be able to create a PR.

Migrate flag is not used

Hi,

On the config documentation you have:

// Disable automatic creation and migration of the outbox table, forcing the application to manage
        // migrations itself
        .migrate(false)

But the flag doesn't seem to do anything. I've looked at the code and here:

  @Override
  public void initialize() {
    if (initialized.compareAndSet(false, true)) {
      try {
        persistor.migrate(transactionManager);
      } catch (Exception e) {
        initialized.set(false);
        throw e;
      }
    }
  }

and here:

  @Override
  public void migrate(TransactionManager transactionManager) {
    DefaultMigrationManager.migrate(transactionManager, dialect);
  }

Probably you are missing the IF condition to check if we want to do the migration or not, or am I missing something?

Instant Null Serialization Bug

Problem:
We are using an Instant as temporal. When it comes to com.gruelbox.transactionoutbox.DefaultInvocationSerializer.InstantTypeAdapter a NullPointerException is thrown.

Expectation:
The NullPointerException is not thrown

Is transaction-outbox stable with postgreSql driver version 14.5.0 ?

We have been upgrading to postgreSql driver version 14.5.0 and maven will link in our specific version instead of the transaction-outbox dependency version.

There is a PR from the dependency-bot to update to 14.5.0 but the tests are failing. Are these tests flaky or is there is actually an issue with transaction-outbox combination and postgreSql driver version 14.5.0?

Supporting load distribution across a cluster

The README states:

that said, if you want to distribute work across a cluster at point of submission, this is also supported

I do, in fact, have that exact requirement but I don't see how to achieve that with the TransactionOutbox API as it stands. How is it supported?

I think you might argue that this functionality is outside the scope of this library, but transaction-outbox is very close to supporting this. All I think it needs is a way to schedule a work item for asynchronous processing by the background thread(s), and not immediate execution in a post-commit hook. This could be done by adding an extra flag to ParameterizedScheduleBuilder.

Is this a feature you would consider adding? I might be able to spin up a PR if so.

ThreadPoolExecutor seems to be single thread by default.

After looking at the code I think you misspelled Math.min instead of Math.max:

/**
   * Schedules background worh with a {@link ThreadPoolExecutor}, sized to match {@link
   * ForkJoinPool#commonPool()} (or one thread, whichever is the larger), with a maximum queue size
   * of 16384 before work is discarded.
   *
   * @return The submitter.
   */
  static Submitter withDefaultExecutor() {
    // JDK bug means this warning can't be fixed
    //noinspection Convert2Diamond
    return withExecutor(
        new ThreadPoolExecutor(
            1,
            Math.min(1, ForkJoinPool.commonPool().getParallelism()),
            0L,
            TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<Runnable>(16384)));
  }

Can you please confirm.

Support printing migration steps

It would be nice to be able to print the migration steps required in case you are disabling the automatic migration functionality of DefaultPersistor because you want to use an existing migration tool like e.g. Flyway.

I assume the requirements could be rather simple: Require the caller to supply a Dialect and then output the resulting SQL with some comments. It could be left to the caller to figure out if they need to start from a specific version.

I could contribute something in case the requirements are pretty clear.

I currently use something locally that that prints the output below by accessing DefaultMigrationManager via reflection:

-- Source: com.gruelbox.transactionoutbox.DefaultMigrationManager

-- 1: Create outbox table
CREATE TABLE TXNO_OUTBOX (
    id VARCHAR(36) PRIMARY KEY,
    invocation TEXT,
    nextAttemptTime TIMESTAMP(6),
    attempts INT,
    blacklisted BOOLEAN,
    version INT
);

-- 2: Add unique request id
ALTER TABLE TXNO_OUTBOX ADD COLUMN uniqueRequestId VARCHAR(100) NULL UNIQUE;

-- 3: Add processed flag
ALTER TABLE TXNO_OUTBOX ADD COLUMN processed BOOLEAN;

-- 4: Add flush index
CREATE INDEX IX_TXNO_OUTBOX_1 ON TXNO_OUTBOX (processed, blacklisted, nextAttemptTime);

-- 5: Increase size of uniqueRequestId
ALTER TABLE TXNO_OUTBOX ALTER COLUMN uniqueRequestId TYPE VARCHAR(250);

-- 6: Rename column blacklisted to blocked
ALTER TABLE TXNO_OUTBOX RENAME COLUMN blacklisted TO blocked;

-- 7: Add lastAttemptTime column to outbox
ALTER TABLE TXNO_OUTBOX ADD COLUMN lastAttemptTime TIMESTAMP(6);

-- 8: Update length of invocation column on outbox for MySQL dialects only.
-- Nothing for PostgreSQL

Provide an interface for generating sequences

We have value generation going through Redis. Is it possible to add an interface so that sequence generation can be done in different ways?

java.lang.reflect.UndeclaredThrowableException

Hello,

I'm running into an issue with the following scenario:

• publishing (and then persisting) twice the same message (with same uniqueRequestId)
  • I always get and UndeclaredThrowableException exception.

java.lang.reflect.UndeclaredThrowableException: null
	at com.gruelbox.transactionoutbox.spring.$Proxy215.executeUpdate(Unknown Source)
	at com.gruelbox.transactionoutbox.DefaultPersistor.save(DefaultPersistor.java:124)
	at com.gruelbox.transactionoutbox.TransactionOutboxImpl.lambda$schedule$17(TransactionOutboxImpl.java:234)
	at com.gruelbox.transactionoutbox.spi.Utils.uncheckedly(Utils.java:50)
	at com.gruelbox.transactionoutbox.TransactionOutboxImpl.lambda$schedule$18(TransactionOutboxImpl.java:219)
	at com.gruelbox.transactionoutbox.spi.ProxyFactory.lambda$constructProxy$2(ProxyFactory.java:87)
	at com.test.newapp.rabbit.out.dummy.DummyUpdatedEventRabbitMqPublisherImpl$ByteBuddy$WihNmQUu.publish(Unknown Source)
	at com.test.newapp.app.configurations.transactionoutbox.TransactionOutBoxPublishersConfiguration$OutBoxDummyUpdatedEventPublisher.publish(TransactionOutBoxPublishersConfiguration.java:51)
	at com.test.newapp.app.configurations.transactionoutbox.OutBoxController.publish(OutBoxController.java:98)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:351)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:196)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:765)
	at org.springframework.transaction.interceptor.TransactionInterceptor$1.proceedWithInvocation(TransactionInterceptor.java:123)
	at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:385)
	at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:119)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184)
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:765)
	at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:717)
	at com.test.newapp.app.configurations.transactionoutbox.OutBoxController$$SpringCGLIB$$0.publish(<generated>)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:259)
	at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:192)
	at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:118)
	at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:920)
	at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:830)
	at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)
	at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1089)
	at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:979)
	at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1014)
	at org.springframework.web.servlet.FrameworkServlet.doPost(FrameworkServlet.java:914)
...
Caused by: java.lang.reflect.InvocationTargetException: null
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:118)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at com.gruelbox.transactionoutbox.spring.SpringTransactionManager$BatchCountingStatementHandler.invoke(SpringTransactionManager.java:148)
	... 84 common frames omitted
Caused by: org.postgresql.util.PSQLException: ERROR: duplicate key value violates unique constraint "txno_outbox_uniquerequestid_key"
  Detail: Key (uniquerequestid)=([email protected]) already exists.
	at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2713)
	at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2401)
	at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:368)
	at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:498)
	at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:415)
	at org.postgresql.jdbc.PgPreparedStatement.executeWithFlags(PgPreparedStatement.java:190)
	at org.postgresql.jdbc.PgPreparedStatement.executeUpdate(PgPreparedStatement.java:152)
	at com.zaxxer.hikari.pool.ProxyPreparedStatement.executeUpdate(ProxyPreparedStatement.java:61)
	at com.zaxxer.hikari.pool.HikariProxyPreparedStatement.executeUpdate(HikariProxyPreparedStatement.java)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)

Unblock all blocked entries

Hello,

There is a hook for blocked entries through the listener, and also TransactionOutbox.unblock(entryId) method.

Have you considered to add :

• a service that unblocks all blocked entries
• a service that exposes blocked entries

Versioning on invocations serialized to the database

Hello,

I ran into an issue with the following scenario:

1. A record gets written to the outbox table in the database.
2. The record fails to process but remains unblocked.
3. New code is deployed that will expect something different with the invocation column of a record in the database (e.g., an argument now contains field B instead of field A).
4. The old record is attempted to be processed using the new code but can no longer be deserialized due to the changes deployed in the latest code.

Based on the above scenario, I am ultimately wondering if there is any way currently to support some form of versioning on records that get written to the database and how those older records would be processed if new code was deployed.

As a side note related to this, I noticed that if deserialization of a record does fail in the flush call, it looks like an exception is thrown, which will halt any further processing of records in that same flush call. Is there a way to skip processing of records that will throw an exception in the flush call. Otherwise, it seems like any processing of records would be halted until the issue with the specific problem record was resolved.

transactionoutbox-spring depends on specific Spring version

At the moment, it depends on Spring 5.2.6, but there's no guarantee that this will be the version that is run by the consumer.

Shouldn't the spring dependency (and others like it) be set to <scope>provided/<scope>?

Docs: Dependency Scopes

That way the testing will be done against a specific version, but it's pulled in by consumers of the lib, and it's not transitive.

This then behaves a little like an npm "peer dependency".

Or is there a better way to achieve this that I am not aware of?

Connections not closed with SpringTransactionManager

Hi,

thanks for your great library!

I tried it out using the spring-integration. I noticed that db-connections won't be given back to the connection pool if a callback throws an exception. This leads to the situation, that after some failed callbacks no connections are available any more.

The problem seems to be that the default methods of com.gruelbox.transactionoutbox.TransactionManager are not wrapped by Springs TransactionInterceptor, because they are not directly annotated with @transactional.

Example: Method inTransactionThrows called in TransactionOutboxImpl.updateAttemptCount is not wrapped by Springs transaction-aspect and thus connections are not given back to the connection pool after the method ends. inTransactionThrows calls inTransactionReturnsThrows, which is not intercepted by the aop-proxy when called from within the object, only when called directly.

Copying all default methods to SpringTransactionManager and annotating them with @transactional solves the problem in my test setup.

Example:

@beta
@service
public class SpringTransactionManager implements ThreadLocalContextTransactionManager {
...
@OverRide
@transactional(propagation = Propagation.REQUIRES_NEW)
public void inTransaction(Runnable runnable) {
uncheck(() -> inTransactionReturnsThrows(ThrowingTransactionalSupplier.fromRunnable(runnable)));
}

@Override
@Transactional(propagation = Propagation.REQUIRES_NEW)
public void inTransaction(TransactionalWork work) {
	uncheck(() -> inTransactionReturnsThrows(ThrowingTransactionalSupplier.fromWork(work)));
}

....

Best regards
Oliver

DefaultInvocationSerializer does not allow java.util.UUID

java.util.UUID is a commonly-used JDK class and is commonly used to model surrogate identifiers. It is therefore a good candidate to be passed as an argument to "scheduled" invocations with transaction outbox.

DefaultInvocationSerializer does not currently appear to allow it. There are various work-arounds, but it seems like something that should be allowed. Do you agree? If so I will raise a PR.

Cache Objenesis ObjectInstantiators

Performance when Objenesis gets involved could be improved by caching the instantiators.

Question: what is the reason for using GSON in DefaultInvocationSerializer?

Hello,
Why GSON is used in DefaultInvocationSerializer? I understand this component can be overridden and I'm planning to rewrite it using Jackson for my spring boot app.
But you did choose GSON there is some reason I think, could you tell about this, please?

Btw about my problem with GSON:
I must define all classes which can be de/serialized - it's not a good solution when you have a lot of classes. In my case it's over 100 classes and this list is not final. We can have and update this list with classes, but it's weird. We don't need it with Jackson.
All my classes extends one and I got all needed classes using reflection. In my case it's not a big problem.

Different problems while migrating to Java 16/17 because of --illegal-access=deny by default

Because of https://openjdk.java.net/jeps/403 several problems appear while using java 16/17:

• CGLIB has been destroyed completely, so now it isn't possible to use inheritance proxies. Looks like this library isn't in an active development state (the latest version of it was released in 2019) despite the cglib develovers know about these problems cglib/cglib#191
  It's still possible to use interface-based proxies though.

• there is a problem with serializing/deserializing classes from java.time with the DefaultInvocationSerializer (based on gson). Same reason - this module is not open.

• I can't actually build this project on openjdk 17 - something with the deprecated com.sun. packages (haven't dig into it tbh).

That's what I found atm. If I find sth else, I'll post it here too.

Bubble up the error from within the scheduled rest request

Hello, I want to throw the error from my scheduled request when making a rest call. I can see it in the logs, but it returns a 200 as the commit overrides it. It is nice that it will get retried before getting blocked, but I need to let the browser know how the request went.

Is there an easy way to achieve this?

Thanks!

Consider downgrading compiler target to 1.8

There may be potential users who are still stuck on such an old JVM. If there's any interest in downgrading, it won't be particularly hard to strip out the Java 9/10/11 features like var and I've no real objection to doing so.

Block topic rows after maxAttempts

In the current implementation of "ordered" tasks will not be blocked, but will be processed indefinitely. maybe it's worth blocking all tasks in a topic if the first task was not completed?

Java Module system compatibility

Hi

Thanks for this project. We think it looks promising and would like to try it out in one of our projects.
However, we use the Java module system and we are not able to require transactionoutbox-core and transactionoutbox-spring at the same time.

All subproject artifacts contain modules that export to the same package com.gruelbox.transactionoutbox.
I suppose this is default behaviour when the same package structure is used throughout the project.

Split packages are not allowed by the Java module system.
Would it be possible to alter the package structure in every subproject so that they are more specific?

Non-blocking support (R2DBC)

Question: Order of outbox entry processing

I can't find any documentation about the 'ordering' of outbox entries that will be processed.

Sometimes it's desirable that the order in which outbox entries are created (for example in case of 'events'), is the same order in which outbox entries are provided for submission (via submitter). This can be hard to achieve when running multiple instances of the same application and would require locking on database level.

Is this library able to achieve same outbox-entry-creation outbox-entry-submission ordering semantics?
If so, does this apply for running multiple instances?

Thanks in advance.

Ruben

Problems with jakarta 3 dependencies

Jakarta 3 brings us a new package structure replacing javax.* with jakarta.*.
As I understand it, in transaction-outbox project, this migration had happened while hibernate validator was updated from 6 to 7. So now, we have a lot of jakarta.* imports across the project. This causes a lot of troubles because a vast majority of libraries and frameworks (e.g. the newest spring boot) still use jakarta 2 with javax.* imports, so you need to use an older version of transaction-outbox in this case to make the code compile.
Is it possible to do something with it? We can downgrade hibernate validator or keep two versions of transaction-outbox? Or even refactor classes with potential problems to make them independent from it.

TransactionOutboxImpl#scheduler can't be shutdown without stopping the whole JVM

Hello,

Our application is deployed as a WAR in a standalone Tomcat. The application can 'reboot' without rebooting the whole JVM.
It seems there is no way to programmatically stop TransactionOutboxImpl#scheduler or to provide it.

Could you please make TransactionOutbox closeable with a close method shutting down the executor?

Wrong duration format in idempotency protection log

The message "Expired idempotency protection on {} requests completed more than {} ago" shows incorrect duration - all values are "shifted" - e.g. "1d:00h:00m" when retention threshold is 1 hour.

Invocation args are serialized as Json Object instead of array

See https://github.com/gruelbox/transaction-outbox/blob/master/transactionoutbox-jackson/src/main/java/com/gruelbox/transactionoutbox/jackson/CustomInvocationSerializer.java

They should be serialized like parameterTypes are ?

Support collections in method arguments

The reason for only supporting arrays and not collections is historical and can be fixed now.

Release 5.2.337 not present on Maven Central

Problem:

I am wondering how and when release 5.2.337 will be available on Maven Central (https://mvnrepository.com/artifact/com.gruelbox/transactionoutbox-core) since https://github.com/gruelbox/transaction-outbox/actions/runs/4383418970 failed.

Expectation:

Release 5.2.337 is available on Maven Central.

High Availability and clustering

If there are multiple instances of the same app connected to the same db and use the same txno_outbox table, would that mean that there can be some sort of race condition between instances with outbox used or it cannot happen because only one thread has knowledge about id of that row? If we want to avoid that do we need to implement Clustering?