kafkaflow-contrib's Introduction

KafkaFlow - extra libraries

This project contains a set of libraries that contribute to KafkaFlow ecosystem.

KafkaFlow.Contrib.Outbox

is a library to provide transactional outbox for KafkaFlow.

The following backends are implemented:
- KafkaFlow.Contrib.Outbox.Postgres - Postgres SQL backend
KafkaFlow.Contrib.ProcessManagers

is a library that provides Process Managers functionality (sometimes also called Sagas).

The following backends are implemented:
- KafkaFlow.Contrib.ProcessManagers.Postgres - Postgres SQL backend for storing process' state

Usage example

As a pattern, Process Managers requires that the state cannot be an eventually-consistent projection, and must be immediately consistent. It also requires that any messages that are published must be transactionally consistent with the state changes.

It means that using process managers implies using Outbox pattern.

Here is how process managers and outbox can be used together:

services
    // We need an NpgsqlDataSource shared between Outbox and Process Managers
    // to be able to update state and send messages transactionally
    .AddSingleton(myNpgsqlDataSource)
    .AddPostgresProcessManagerState()
    .AddPostgresOutboxBackend()
    .AddKafka(kafka =>
        kafka
            .AddCluster(cluster =>
                cluster
                    // The dispatcher service will be started in background
                    .AddOutboxDispatcher(dispatcher =>
                        // I strongly recommend to use Murmur2Random since it is
                        // the "original" default in Java ecosystem, and is shared by other
                        // ecosystems such as JavaScript or Python.
                        dispatcher.WithPartitioner(Partitioner.Murmur2Random))
                    .AddProducer("default", producer =>
                        producer
                            // Make this producer go through the outbox
                            .WithOutbox()
                            .AddMiddlewares(m => m.AddSerializer<JsonCoreSerializer>()))
    // and so on

Recommend Projects