# `Outboxer.Relay.Kafka`
[🔗](https://github.com/epinault/outboxer/blob/v0.1.0/lib/outboxer/relay/kafka.ex#L2)

Relays messages to Kafka through [brod](https://hex.pm/packages/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` (`t: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` (`t:keyword/0`) - brod client configuration (SSL, SASL, ...) for the client Outboxer starts. Ignored without `:endpoints`. The default value is `[]`.

* `:producer_config` (`t: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.

---

*Consult [api-reference.md](api-reference.md) for complete listing*