# `BroadwayKafka.Producer`
[🔗](https://github.com/dashbitco/broadway_kafka/blob/v0.6.1/lib/broadway_kafka/producer.ex#L1)

A Kafka connector for Broadway.

BroadwayKafka can subscribe as a consumer to one or more topics and process streams
of records within the same consumer group. Communication is done through Kafka's
[Consumer API](https://kafka.apache.org/documentation.html#consumerapi) using the
[:brod](https://github.com/klarna/brod/) client.

## Options

  * `:hosts` - Required. A list of host and port pairs, or one string of comma-separated `HOST:PORT` pairs, used
to make the first connection to Kafka. For example: `[localhost: 9092]`,
`[{"kafka-vm1", 9092}, {"kafka-vm2", 9092}]`, `"kafka-vm1:9092,kafka-vm2:9092"`.

  * `:group_id` - Required. A (non-empty) unique string that identifies the consumer group that the producer will join.

  * `:topics` - Required. A list of topics that the producer will subscribe to.

  * `:receive_interval` - The duration (in milliseconds) that the producer waits before making
a request for more messages. The default value is `2000`.

  * `:offset_commit_on_ack` - Tells Broadway to send or not an offset commit request after each acknowledgement.
Setting this value to `false` can increase performance since commit requests will
respect the `:offset_commit_interval_seconds` option. However, setting long commit
intervals might lead to a large number of duplicated records to be processed after
a server restart or connection loss. If that's the case, make sure your logic is
**idempotent** when consuming records to avoid inconsistencies. Also, bear
in mind the the negative performance impact might be insignificant if you're using
batchers since only one commit request will be performed per batch. The default value is `true`.

  * `:offset_reset_policy` - The offset to use when Kafka has no saved offset or the saved offset has expired. Use
`:earliest`, `:latest`, or `{:timestamp, timestamp}`, where `timestamp` is a non-negative
time in milliseconds. The default value is `:latest`.

  * `:begin_offset` - How consumers choose their first offset. When set to `:assigned`, the starting offset
will be the one returned in the Kafka partition assignments (the latest committed
offsets for the consumer group). When set to `:reset`, the starting offset will be
dictated by the `:offset_reset_policy` option, either starting from the `:earliest`
or the `:latest` offsets of the topic. The default value is `:assigned`.

  * `:shared_client` - When `false`, it starts one `:brod` client per producer. When `true`, it starts a
single shared `:brod` client across all producers (which may reduce
memory/resource usage). May cause severe performance degradation, see
["Shared Client Performance"](#module-shared-client-performance) for details. The default value is `false`.

  * `:group_config` - Options passed to `:brod`'s group coordinator. The default value is `[]`.

  * `:fetch_config` - The available options to configure how messages are fetched. Unless noted
otherwise, they are internally passed to `:brod.fetch/5`. The default value is `[]`.

  * `:client_config` - Options passed to `:brod.start_client/3` when it starts a client. The default value is `[]`.

### Group config

  * `:group_instance_id` - A unique, non-empty string that identifies this consumer group member across restarts.
This enables [static group membership](https://kafka.apache.org/39/design/design/#static-membership)
and requires `:brod` 4.6.3 or later. When Kafka fences a static member, BroadwayKafka
stops that member instead of trying to take the ID back. Retrying can make old and new
instances fence each other during a rolling deploy and cause repeated group rebalances.
*Available since v0.6.0*.

  * `:offset_commit_interval_seconds` - The time *in seconds* between two `OffsetCommitRequest` messages. The default value is `5`.

  * `:rejoin_delay_seconds` - The delay *in seconds* before rejoining the group. The default value is `1`.

  * `:session_timeout_seconds` - The time *in seconds* that the group coordinator waits before it considers a member "down"
(if no heartbeat or any kind of request is received). A group member may also consider
the coordinator "down" if it receives no heartbeat response within this time. The default value is `30`.

  * `:heartbeat_rate_seconds` - The time *in seconds* between heartbeats ("ping" requests) sent to the group coordinator.
Heartbeats are used to ensure that the consumer's session stays active and
to facilitate rebalancing when new consumers join or leave the group.
The value must be set lower than `:session_timeout_seconds`, typically equal to or lower
than ⅓ of that value. It can be adjusted even lower to control the expected time
for normal rebalances. The default value is `5`.

  * `:rebalance_timeout_seconds` - The time *in seconds* that each worker has to join the group after a rebalance starts.
If the timeout is exceeded, then the worker will be removed from the group,
which will cause offset commit failures. The default value is `30`.

### Fetch config

  * `:min_bytes` - The minimum amount of data to be fetched from the server.
If not enough data is available the request will wait for that much data to accumulate
before answering. Setting this value greater than `1` can improve
server throughput a bit at the cost of additional latency. The default value is `1`.

  * `:max_bytes` - The most data to fetch from one partition at a time. A larger value may improve throughput
at the cost of more memory use. The default value is `1048576`.

  * `:max_wait_time` - The most time (in millisecond) that the broker may wait for `:min_bytes` of data. The default value is `1000`.

  * `:max_fetch_retries` - How many times a failed fetch is retried in place when Kafka returns a retriable error,
before the producer raises as it does for any other error. Set to `0` to disable retries.
BroadwayKafka uses this option itself and does not pass it to `:brod.fetch/5`. The default value is `3`.

  * `:fetch_retry_backoff_ms` - The time, in milliseconds, to wait before the first retry of a failed fetch
(see `:max_fetch_retries`). The wait time doubles on each later retry. BroadwayKafka uses
this option itself and does not pass it to `:brod.fetch/5`. The default value is `500`.

### Client config

  * `:client_id_prefix` - A string added to the client ID that BroadwayKafka builds for `:brod`.

  * `:sasl` - A tuple of "mechanism" (`:plain`, `:scram_sha_256`, or `:scram_sha_512`),
username, and password. See `:brod`'s
[`Authentication Support`](https://github.com/kafka4beam/brod#authentication-support)
documentation for more information. You may also pass `{:callback, module, options}`
for a SASL plug-in. Set this to `:undefined` to disable SASL.

  * `:ssl` - A boolean or a keyword list of SSL/TLS client options. See the
[`tls_client_option`](http://erlang.org/doc/man/ssl.html#type-tls_client_option)
documentation for more information.

  * `:connect_timeout` - The time (in milliseconds) allowed for a connection to Kafka. Default is what
`:brod` defaults to (5s at the time of writing).

  * `:request_timeout` - The time (in milliseconds) allowed for a response from Kafka. It must be at least `1_000`.
Default is to use `:brod`'s default timeout which is currently 4 minutes (`240_000`).

  * `:query_api_versions` - Whether to ask Kafka which API versions it supports when a connection starts. Set this to
`false` for Kafka versions before 0.10.

  * `:extra_sock_opts` - Extra `gen_tcp` socket options. [More info](https://www.erlang.org/doc/man/gen_tcp.html#type-option).
Set to `[:inet6]` if your Kafka broker uses IPv6.

  * `:allow_topic_auto_creation` - Whether `:brod` may send metadata requests that can create a topic when the broker allows
automatic topic creation.

> #### `:brod` Option Support {: .info}
> Currently, Broadway does not support all options provided by `:brod`. If you
> have a scenario where you need any extra option that is not listed above, please open an
> issue, so we can consider adding it.

## Example

    Broadway.start_link(MyBroadway,
      name: MyBroadway,
      producer: [
        module: {BroadwayKafka.Producer, [
          hosts: [localhost: 9092],
          group_id: "group_1",
          topics: ["test"],
        ]},
        concurrency: 1
      ],
      processors: [
        default: [
          concurrency: 10
        ]
      ]
    )

## Concurrency and partitioning

The concurrency model provided by Kafka is based on partitioning, i.e., the more partitions
you have, the more concurrency you get. However, in order to take advantage of this model
you need to set up the `:concurrency` options for your processors and batchers accordingly. Having
less concurrency than topic/partitions assigned will result in individual processors handling more
than one partition, decreasing the overall level of concurrency. Therefore, if you want to
always be able to process messages at maximum concurrency (assuming you have enough resources
to do it), you should increase the concurrency up front to make sure you have enough
processors to handle the extra messages received from new partitions assigned.

> **Note**: Even if you don't plan to add more partitions to a Kafka topic, your pipeline can still
receive more assignments than planned. For instance, if another consumer crashes, the server
will reassign all its topic/partition to other available consumers, including any Broadway
producer subscribed to the same topic.

## Handling failed messages

`BroadwayKafka` never stops the flow of the stream, i.e. it will **always ack** the messages
even when they fail. Unlike queue-based connectors, where you can mark a single message as failed.
In Kafka that's not possible due to its single offset per topic/partition ack strategy. If you
want to reprocess failed messages, you need to roll your own strategy. A possible way to do that
is to implement `c:Broadway.handle_failed/2` and send failed messages to a separated stream or queue for
later processing.

## Message metadata

When producing messages, the following information will be passed to
[`Broadway.Message`](`t:Broadway.Message.t/0`)'s metadata.

  * `topic` - The topic the message was published.

  * `partition` - The topic partition.

  * `offset` - The offset assigned to the message inside the partition.

  * `key` - The partition key.

  * `ts` - A timestamp associated with the message.

  * `headers` - The headers of the message.

## Telemetry

This producer emits a few [Telemetry](https://github.com/beam-telemetry/telemetry)
events which are listed below.

  * `[:broadway_kafka, :assignments_revoked, :start | :stop | :exception]` spans -
    these events are emitted in "span style" when receiving assignments revoked call from consumer group coordinator
    See `:telemetry.span/3`.

  * `[:broadway_kafka, :fenced_instance_id]` - emitted after a producer stops consuming
    because Kafka fenced its static group member. The measurement is `:system_time`. The
    metadata includes `:producer`, `:client_id`, `:group_id`, and `:group_instance_id`.

## Shared Client Performance

Enabling shared client may drastically decrease performance. Since connection is handled by a single process,
producers may block each other waiting for the client response.

This is more likely to be an issue if the producers on your pipeline are fetching message from
multiple topics and specially if there are very low traffic topics, which may block on batch wait times.

To mitigate this, you can split your topics between multiple pipelines, but notice that this will
increase the resource usage as well. By creating one new client/connection for each pipeline,
you effectively diminishing the `shared_client` resource usage gains. So make sure to measure
if you enable this option.

---

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