Add a message bus for transactional outbox deliveries

[errm]

A service that updates database entities, and also writes a related event to kafka can have some issues if we decide to wrap the database updates in a transaction, then some error or delay occurs when writing and event to kafka. Especially if the event is written synchronously.

This can also cause lock conntention and increased resource useage on the database server at scale.

• https://microservices.io/patterns/data/transactional-outbox.html
• https://docs.aws.amazon.com/prescriptive-guidance/latest/cloud-design-patterns/transactional-outbox.html

One solution is to write the event stream to a dedicated database table, and have an additonal process to handle writing the events to kafka.

This means that the application doesn't need to manage it's own connections to kafka, and transactions can be used in the normal way without any downsides or performance degredation.

This change provides a new OutboxMessageBus that can be configured with an active record model.

e.g.

migration:

class CreateKafkaOutboxEvents < ActiveRecord::Migration[7.0]
  def change
    create_table :kafka_outbox_events do |t|
      t.string :topic
      t.string :key
      t.column :payload, :longblob # for avro - text would be more
appropriate for JSON
      t.timestamps
    end

    add_index :kafka_outbox_events, :topic
  end
end

config/initializers/streamy.rb:

require "streamy/message_buses/outbox_message_bus"
class KafkaOutboxEvent < ActiveRecord::Base; end
Streamy.message_bus = Streamy::MessageBuses::OutboxMessageBus.new(model: KafkaOutboxEvent)

This implimentation only allows for the use of a single table as the outbox. If we wanted to e.g. use a table per topic then the implimentation will need to be a bit more complex.

For now, I suspect that indexing on the topic collum will be good enough, as we can run multiple consuming workers each selecting a different topic concurrently.

We will only be able to use a single worker to select rows (with locking) in the consuming process where the backend is MySQL 5.7 however with an upgrade to MySQL 8+ we can make use of SKIP LOCKED to increase concurrency if required.

[robertomiranda]

We will only be able to use a single worker to select rows (with locking) in the consuming process where the backend is MySQL 5.7 however with an upgrade to MySQL 8+ we can make use of SKIP LOCKED to increase concurrency if required.

Make sense!

Looking good, thanks @errm!

[shimpeko]

I've been sceptical about data integrity between Kafka and DB (ex. DB commit can fail after writing to Kafka). Great improvement.