Broadway Kafka

https://hexdocs.pm/broadway_kafka
Admin
Broadway Kafka is a Kafka connector for Broadway that enables processing streams of records from one...
Tokens:3,957
Snippets:61
Trust Score:9
Update:5 months ago
Show doc for...
Tokens:
Raw
### Start Broadway Kafka Producer

Source: https://hexdocs.pm/broadway_kafka/index.html

This example shows how to start a Broadway pipeline with a Kafka producer. Ensure the producer module, hosts, group ID, and topics are correctly configured.

```elixir
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
    ]
  ]
)
```

--------------------------------

### Kafka Client ID Prefix Example

Source: https://hexdocs.pm/broadway_kafka/BroadwayKafka.Producer.html

Example of a client_id_prefix configuration for `:brod.start_client/3`. This helps in identifying the client in connection logs.

```elixir
client_id_prefix: :"#{Node.self()} -"
```

--------------------------------

### Kafka Client Connection Log Example

Source: https://hexdocs.pm/broadway_kafka/index.html

Illustrates a typical connection log for a Kafka client using `:client_id_prefix`. This helps in understanding how client IDs are generated and logged.

```elixir
20:41:37.717 [info]      :supervisor: {:local, :brod_sup}
:started: [
  pid: #PID<0.286.0>,
  id: :"nonode@nohost - Elixir.BroadwayKafka.ConsumerTest.MyBroadway.Broadway.Producer_0.Client",
  mfargs: {:brod_client, :start_link,
   [
     [localhost: 9092],
     :"nonode@nohost - Elixir.BroadwayKafka.ConsumerTest.MyBroadway.Broadway.Producer_0.Client",
     [client_id_prefix: :"nonode@nohost - "]
   ]},
  restart_type: {:permanent, 10},
  shutdown: 5000,
  child_type: :worker
]
```

--------------------------------

### Kafka Host Connection Formats

Source: https://hexdocs.pm/broadway_kafka/index.html

Specifies the different formats for providing Kafka host connection details. Ensure the format matches your Kafka cluster setup.

```elixir
# Keyword
["kafka-vm1": 9092, "kafka-vm2": 9092, "kafka-vm3": 9092]
```

```elixir
# List of tuples
[{"kafka-vm1", 9092}, {"kafka-vm2", 9092}, {"kafka-vm3", 9092}]
```

```elixir
# String
"kafka-vm1:9092,kafka-vm2:9092,kafka-vm3:9092"
```

--------------------------------

### Kafka Host Connection Options

Source: https://hexdocs.pm/broadway_kafka/BroadwayKafka.Producer.html

Specifies how to provide Kafka host connection details. Use a keyword list, a list of tuples, or a comma-separated string.

```elixir
["kafka-vm1": 9092, "kafka-vm2": 9092, "kafka-vm3": 9092]
```

```elixir
[{"kafka-vm1", 9092}, {"kafka-vm2", 9092}, {"kafka-vm3", 9092}]
```

```elixir
"kafka-vm1:9092,kafka-vm2:9092,kafka-vm3:9092"
```

--------------------------------

### BroadwayKafka.Producer Group Config Options

Source: https://hexdocs.pm/broadway_kafka/index.html

Configuration options for the Kafka group coordinator used by BroadwayKafka.

```APIDOC
## BroadwayKafka.Producer Group Config Options

### Description
Options passed to `:brod`'s group coordinator.

### Parameters
#### Optional Parameters
- **`:offset_commit_interval_seconds`** (integer) - Optional. The time interval between two OffsetCommitRequest messages. Default is 5.
- **`:rejoin_delay_seconds`** (integer) - Optional. Delay in seconds before rejoining the group. Default is 1.
- **`:session_timeout_seconds`** (integer) - Optional. Time in seconds the group coordinator broker waits before considering a member 'down' if no heartbeat or any kind of request is received. Default is 30 seconds.
- **`:heartbeat_rate_seconds`** (integer) - Optional. Time in seconds for member to 'ping' group coordinator. Default is 5 seconds.
- **`:rebalance_timeout_seconds`** (integer) - Optional. Time in seconds for each worker to join the group once a rebalance has begun. Default is 30.
```

--------------------------------

### BroadwayKafka.Producer Options

Source: https://hexdocs.pm/broadway_kafka/index.html

Configuration options for the BroadwayKafka Producer. These options control how the producer connects to Kafka, subscribes to topics, and manages message consumption.

```APIDOC
## BroadwayKafka.Producer Options

### Description
Configuration options for the BroadwayKafka Producer.

### Parameters
#### Required Parameters
- **`:hosts`** (list or string) - Required. A list of host and port tuples or a single string of comma separated HOST:PORT pairs to use for establishing the initial connection to Kafka, e.g. [localhost: 9092].
- **`:group_id`** (string) - Required. A unique string that identifies the consumer group the producer will belong to.
- **`:topics`** (list) - Required. A list of topics that the producer will subscribe to.

#### Optional Parameters
- **`:receive_interval`** (integer) - Optional. The duration (in milliseconds) for which the producer waits before making a request for more messages. Default is 2000 (2 seconds).
- **`:offset_commit_on_ack`** (boolean) - Optional. Tells Broadway to send or not an offset commit request after each acknowledgement. Default is `true`.
- **`:offset_reset_policy`** (atom or tuple) - Optional. Defines the offset to be used when there's no initial offset in Kafka or if the current offset has expired. Possible values are `:earliest`, `:latest` or `{:timestamp, timestamp}` (in milliseconds). Default is `:latest`.
- **`:begin_offset`** (atom) - Optional. Defines how to get the initial offset for the consumers. The possible values are `:assigned` or `:reset`. Default is `:assigned`.
- **`:shared_client`** (boolean) - Optional. When false, it starts one client per producer. When true, it starts a single shared client across all producers. Default is `false`.
- **`:group_config`** (list) - Optional. A list of options used to configure the group coordinator.
- **`:fetch_config`** (list) - Optional. A list of options used when fetching messages.
- **`:client_config`** (list) - Optional. A list of options used when creating the client.
```

--------------------------------

### Update BroadwayKafka Producer Topics

Source: https://hexdocs.pm/broadway_kafka/BroadwayKafka.html

Sequentially updates topics for all Broadway producers in a given pipeline. Call this function when you need to change the topics a producer is subscribed to.

```elixir
BroadwayKafka.update_topics(MyBroadway, ["topic_a", "topic_b"])
:ok

BroadwayKafka.update_topics(MyBroadway, [])
:ok
```

--------------------------------

### update_topics

Source: https://hexdocs.pm/broadway_kafka/BroadwayKafka.html

Sequentially updates topics in all Broadway producers within a specified pipeline. This function allows dynamic reconfiguration of Kafka topics without restarting the pipeline.

```APIDOC
## update_topics(name, topics)

### Description
Sequentially updates topics in all Broadway producers in the pipeline given by `name`.

### Parameters
#### Path Parameters
- **name** (string) - Required - The name of the Broadway pipeline.
- **topics** (list) - Required - A list of topic names to update.

### Request Example
```elixir
BroadwayKafka.update_topics(MyBroadway, ["topic_a", "topic_b"])
```

### Response
#### Success Response (200)
- `:ok` - Indicates the topics were updated successfully.

### Response Example
```elixir
:ok
```
```
Tokens:
Raw