Outboxer.Relay.Kafka (Outboxer v0.1.0)

Relays messages to Kafka through brod.

This module only exists when :brod is in your dependencies:

{:brod, "~> 4.0"}

With a brod client you already run

{Outboxer,
 repo: MyApp.Repo,
 outboxes: [
   orders: [relay: {Outboxer.Relay.Kafka, client: :my_brod_client, topic: "orders"}]
 ]}

Letting Outboxer start the client

Give :endpoints and Outboxer supervises a brod client under the outbox, registered as :client:

orders: [
  relay:
    {Outboxer.Relay.Kafka,
     client: :outboxer_orders,
     endpoints: [{"localhost", 9092}],
     topic: "orders",
     producer_config: [required_acks: -1, compression: :snappy]}
]

Options

:client (atom/0) - Required. The brod client id. An existing client, or the name to register the one Outboxer starts.
:topic - Required. The topic: a string, a function taking the envelope, or {module, function, extra_args} called with the envelope first. A function must send a given group to a single topic.
:endpoints - Kafka bootstrap endpoints, such as [{"localhost", 9092}]. When set, Outboxer starts and supervises the brod client. The default value is nil.
:client_config (keyword/0) - brod client configuration (SSL, SASL, ...) for the client Outboxer starts. Ignored without :endpoints. The default value is [].
:producer_config (keyword/0) - brod producer configuration, used when a topic's producer is started. The default value is [required_acks: -1].
:partitioner - :hash hashes the group key over the topic's partitions. Or a function taking the envelope and the partition count, returning a partition. It must be deterministic for a group. The default value is :hash.

How messages map to Kafka

Kafka	Outboxer
topic	the `:topic` option
key	the message's group key (empty when it has none)
value	the encoded payload, as is
headers	the message headers, plus `outboxer-id` and `outboxer-type`
partition	hash of the key; random for messages without a key

Ordering

A group key becomes the Kafka key, so a group always goes to the same Kafka partition and Kafka keeps the order Outboxer delivered in. A batch is produced with one synchronous request per topic-partition and Outboxer waits for the acknowledgement before it removes the messages.

The partition is chosen by Outboxer, not by brod's partitioner, and it depends on the topic's partition count. If you add partitions to a topic, keys move, as they would with any hash partitioner.

Use required_acks: -1 (the default here) unless you can afford to lose acknowledged messages when a broker fails.

Duplicates

Delivery is at-least-once. If the broker stored a batch but the acknowledgement was lost, or the database commit that follows fails, the batch is produced again. Consumers can deduplicate on the outboxer-id header, which is stable across attempts.