Add support for async completion of Outbox entries

[chychkan]

First of all, I really appreciate that this library exists, its design and its simplicity. So, thank you! ❤️

Problem

We’d like to use this library to publish messages to Kafka. The KafkaProducer#send() API is fully asynchronous: it only enqueues the record into the client’s internal buffers and returns immediately. The message is not considered delivered until the returned Future completes or the callback is invoked.

Because of this, calling send() inside an outbox-invoked method does not mean the message has been delivered (or even attempted). If the JVM or producer crashes after send() returns but before Kafka acknowledges the write, the message is lost - even though the Outbox will mark the entry as “processed”.

This breaks the main guarantee of the Outbox pattern: only mark the entry as processed when the external system has safely accepted it.

Kafka requires waiting for the Future (or callback) to know whether the send succeeded or failed.

Impact

The only way to ensure delivery today is to call producer.flush() after every message, but this significantly reduces the throughput. And so the library becomes usable only for low-throughput use cases.

Potential solutions

It would help if the proxied method could:

1. return a Future / CompletableFuture, which the Outbox would wait on before marking the entry as processed
   or
2. accept a callback supplied by the Outbox, which the user could invoke when the asynchronous operation completes

This would allow integrating properly with APIs like Kafka’s producer without forcing a flush-per-message.

Question

Has anyone solved this use case? Is there an existing way to handle async completion, or would you be open to supporting one of these approaches?

[badgerwithagun]

I have indeed looked into this before! The design of the library entirely supports this with a small change. I'll have a think of the quickest way.

[badgerwithagun]

Been thinking about this.

Fundamentally the process of handling an outbox job is synchronous because it involves a database transaction:

1. Read and lock the record
2. Process the work (e.g. send to Kafka)
3. Delete or mark the record processed
4. Commit

If instead of treating (2) as blocking, we perform (3) and (4) in a callback, we have to assume that the thread handling the callback will be different to the one handling the commit.

This doesn't play nice with most transaction managers which tend to use threadlocal transaction state, or some JDBC drivers which don't have thread-safe connections.

To be honest, I find myself wondering "why don't you just call KafkaProducer.send(...).join()? OK it's blocking but a DB connection is blocked while waiting even if we don't block a thread.

Furthermore, if you use a virtual thread executor for your Submitter implementation, then the join will actually do a wait and callback on the carrier thread anyway, so the end result will be the same.

In many ways virtual threads have eliminated the need for non-blocking I/O these days, especially for this sort of coroutine/yield type behaviour.

On a phone at the moment but if any of that is unclear I can type out an example when I'm in front of a PC.

[chychkan]

@badgerwithagun thanks for looking into this!

Using KafkaProducer.send(...).join() would be equivalent to invoking KafkaProducer.flush() for each message. It would be totally synchronous and the Outbox would guarantee the delivery, but the throughput would suffer.

The idea with the callback that I mentioned could be something like this:

1. Outbox reads and locks a batch of records (let's say 100 of them)
2. For each of the records Outbox creates a CompletableFuture.
3. Outbox invokes the target method for each of the records and passes to it a callback, which when invoked resolves the CompletableFuture that corresponds to the record.
4. The target method invokes KafkaProducer.send(...) with a callback that resolves the callback provided by Outbox and returns immediately.
5. Outbox sequentially waits on each of the CompletableFuture's created in the same order as the target invocations and as soon as a future is completed, marks the corresponding record as processed.
6. When all futures are completed, Outbox commits the transaction.

What this gives us is that we fetch a batch of 100 records, we invoke the target method for all 100 of them (all 100 of them are added to Kafka's buffer and return immediately, but not delivered yet), then we wait for the completions and mark them "processed" and they may become resolved very soon because the buffer fills up quickly. And then we commit when all 100 of them are delivered. Does this make sense or am I missing something?

[badgerwithagun]

Using KafkaProducer.send(...).join() would be equivalent to invoking KafkaProducer.flush() for each message

I don't think this is true, at least reading the Kafka docs. Each individual CompletableFuture will only complete (causing join to return) when that message is flushed. Calling join doesn't cause a flush.

Imagine here that KafkaProducer is configured to send in batches of 3. It would work like this:

sequenceDiagram
    actor client
    participant virtualThread1
    participant virtualThread2
    participant virtualThread3
    participant kafkaProducer
    participant broker
    client->>+virtualThread1: outbox.schedule(getClass()).send(1)
    client->>+virtualThread2: outbox.schedule(getClass()).send(2)
    client->>+virtualThread3: outbox.schedule(getClass()).send(3)
    virtualThread1->>kafkaProducer: send(...).join();
    activate kafkaProducer
      virtualThread2->>kafkaProducer: send(...).join();
      virtualThread3->>kafkaProducer: send(...).join();
      kafkaProducer->>broker : Reached three messages.<br>Send batch
      kafkaProducer-->>virtualThread1: notify()
      kafkaProducer-->>virtualThread2: notify()
      kafkaProducer-->>virtualThread3: notify()
    deactivate kafkaProducer
    virtualThread1->>-virtualThread1: Mark job processed and commit
    virtualThread2->>-virtualThread2: Mark job processed and commit
    virtualThread3->>-virtualThread3: Mark job processed and commit

Loading

Note that the producer will only flush when it gets enough messages, and then resolve all the waiting CompletableFutures.

Your point about performance is worth a bit more though though. The only limitations are going to be from either your thread pool or database connection pool.

The first can be eliminated with Submitter.withExecutor(Executors.newVirtualThreadPerTaskExecutor()).

The second is likely to be an issue. If you are firing jobs that quickly, then your database connection pool is already your limiting performance factor; it sets a ceiling on the most jobs that can be processed at once. By making each request wait for KafkaProducer to fill its buffer, each job will last longer which will place more contention on the connection pool.

There's also a deadlock scenario: let's say you have 50 connections in your pool and you need 100 to fill a batch for Kafka: that means the batches will only ever flush through timing out, never through hitting their capacity, so you'll end up limited at only sending messages at a rate of 50 per batch timeout.

What this leads to, and which your solution also deals with, is that we want to avoid holding a database connection while waiting for the response from Kafka. That way, the flow looks more like this (imagine the activations on the virtual threads represent the length of the database transactions):

sequenceDiagram
    actor client
    participant virtualThread1
    participant virtualThread2
    participant virtualThread3
    participant kafkaProducer
    participant broker
    client->>+virtualThread1: outbox.schedule(getClass()).send(1)
    client->>+virtualThread2: outbox.schedule(getClass()).send(2)
    client->>+virtualThread3: outbox.schedule(getClass()).send(3)
    virtualThread1->>kafkaProducer: send(...).join();
    deactivate virtualThread1
    activate kafkaProducer
      virtualThread2->>kafkaProducer: send(...).join();
      deactivate virtualThread2
      virtualThread3->>kafkaProducer: send(...).join();
      deactivate virtualThread3
      kafkaProducer->>broker : Reached three messages.<br>Send batch
      kafkaProducer-->>+virtualThread1: notify()
      virtualThread1->>-virtualThread1: Mark job processed
      kafkaProducer-->>+virtualThread2: notify()
      virtualThread2->>-virtualThread2: Mark job processed
      kafkaProducer-->>+virtualThread3: notify()
      virtualThread3->>-virtualThread3: Mark job processed
    deactivate kafkaProducer

Loading

Now we have eliminated all database connection pool contention. Effectively we have achieved exactly what you are after without really changing the way outbox works.

This would be simple enough. Change this method so that it opens a separate transaction before the call to the client method to read the entry and then another one after the client method to update the status on the entry.

There is one problem though: because Outbox is no longer actually locking the record in the DB, another thread can now find it and attempt to process it. For unordered jobs, this is kind of a problem, because it means a lot more duplicate calls (they are always possible, but we do try and avoid it).

What I am now wondering is whether it matters for ordered processing... thinky think

[badgerwithagun]

What we could have is a "non locking" mode where processNowInternal doesn't hold a connection/transaction open while processing. That introduces a lot of additional risks, but may be worth it for performance if combined with a few things:

I assume you are using ordered sending, because Kafka.

Then you just need to call producer.flush() at the end of your outbox flush task, so each outbox.flush() causes a load of producer.send()s to fire and block waiting to be flushed. Once done you call producer.flush(), all the futures resolve and you are safe to go to the next iteration.

[chychkan]

Using KafkaProducer.send(...).join() would be equivalent to invoking KafkaProducer.flush() for each message

I don't think this is true, at least reading the Kafka docs. Each individual CompletableFuture will only complete (causing join to return) when that message is flushed. Calling join doesn't cause a flush.

You're correct, calling the join() itself doesn't cause a flush. I meant to say that by calling join() the Outbox would wait for each message delivery before proceeding to the next message.

The first can be eliminated with Submitter.withExecutor(Executors.newVirtualThreadPerTaskExecutor()).

The second is likely to be an issue.

I see. Yeah, it may indeed exhaust the connection pool. So, if we don't use topics, then each message is submitted into the executor for processing. If we don't have many threads in the executor, then our Kakfa buffer does not fill up. And if we increase the number of threads, then we may run out of connections because each thread gets a new transaction. And using a small Kafka buffer makes the communication with Kafka less efficient. Hmm 🤔 Overall, I'd probably try to minimize the impact on the connection pool (as it's used to serve user-facing requests) and would try to shift load into background (relying more on the outbox.flush(), potentially on another JVM, some worker instance).

In our particular case, most likely we'll be using lots of topics (e.g. the entity_type/entity_id as the topic). Which means, if I understand it right, all the processing would be happening in the outbox.flush(), so would be awesome if we could parallelize that one a bit more. Now it fetches a batch of entries (one from each topic) and then processes them sequentially, but it should be totally safe to process them in parallel as they are all from different topics. So, if we could use the same Sumbitter there, we could get the benefits of parallel processing with no risk of overloading the connection pool as all the entries in a batch there use the same transaction and thus the same single connection. Thoughts?

[chychkan]

Oh, I'm realizing, we can't really use the same Connection as it's not thread safe. So, with flush() we run into the same issue.

[chychkan]

So, each entry needs to be processed in a separate thread and needs a separate connection. The only thing we can possibly optimize here is to find a way not to spend time waiting for Kafka's flush. So, back to the callback idea.

Imagine, we have an executor with 5 threads. In the outbox.flush() we fetch a batch of 1000 entries, we submit each one into the Executor. So, there are 5 of them max being processed at a time. The processing involves: grab a connection from the pool, prepare data, do Kafka#send(), register a callback with Kafka, return. So, we'll work through 1000 of them and will get a 1000 of CompletableFutures. Then we wait for all of them, then we commit. We utilize at most 5 + 1 connections at all times (1 connection for the outbox.flush() thread itself + 5 connections for the executor threads). Am I missing something?

[badgerwithagun]

Firstly: virtual threads, not platform threads. Virtual threads are Java's new native way of doing async code since a couple of years ago. They basically remove the need to dedicate a thread to block on something. Java can handle thousands of virtual threads with no overhead. Well worth a read on the subject.

The problem is the connections. We don't want to hold open the connections while waiting, so we need a small change to this library to allow no-lock processing. That will mean there will be no top limit of waiting futures.

If that's not clear, don't worry, I'll put together some examples code along with a fix to support this as soon as I can.

[chychkan]

Sounds good! Thank you! I understand that virtual threads are cheap and I'm not concerned about threads. The connections is the limiting factor indeed.