### Configure and Start Broadway Kafka Pipeline

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

This example demonstrates how to configure and start a Broadway pipeline using BroadwayKafka.Producer. It includes required options like hosts, group_id, and topics, as well as optional settings for polling, offset commits, group coordination, fetch tuning, and client configurations. Ensure all necessary Kafka connection details and consumer group settings are provided.

```elixir
defmodule MyApp.KafkaPipeline do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {BroadwayKafka.Producer, [
          # Required
          hosts: [localhost: 9092],
          group_id: "my_consumer_group",
          topics: ["orders", "payments"],

          # Optional polling/offset settings
          receive_interval: 500,            # ms between polls (default: 2000)
          offset_commit_on_ack: true,       # commit on every ack (default: true)
          offset_reset_policy: :earliest,   # :earliest | :latest | {:timestamp, ms}
          begin_offset: :assigned,          # :assigned | :reset

          # Optional group coordinator config
          group_config: [
            offset_commit_interval_seconds: 5,
            rejoin_delay_seconds: 1,
            session_timeout_seconds: 30,
            heartbeat_rate_seconds: 5,
            rebalance_timeout_seconds: 30
          ],

          # Optional fetch tuning
          fetch_config: [
            min_bytes: 1,
            max_bytes: 1_048_576,   # 1 MiB
            max_wait_time: 1_000
          ],

          # Optional client config
          client_config: [
            ssl: true,
            sasl: {:plain, "username", "password"},
            connect_timeout: 5_000,
            request_timeout: 30_000,
            client_id_prefix: "my_app-"
          ],

          # Optional: share a single brod client across all producers
          shared_client: false
        ]},
        concurrency: 3   # number of producer processes (= max parallel partition groups)
      ],
      processors: [
        default: [concurrency: 10]
      ],
      batchers: [
        default: [
          batch_size: 100,
          batch_timeout: 200,
          concurrency: 5
        ]
      ]
    )
  end

  @impl Broadway
  def handle_message(_processor, message, _context) do
    # message.data       — raw binary payload
    # message.metadata   — %{topic, partition, offset, key, ts, headers}
    IO.inspect(message.metadata, label: "Kafka metadata")
    message
  end

  @impl Broadway
  def handle_batch(_batcher, messages, batch_info, _context) do
    {topic, partition} = batch_info.batch_key
    IO.puts("Batch from #{topic}/#{partition}: #{length(messages)} messages")
    messages
  end
end

# Start the pipeline
{:ok, _pid} = MyApp.KafkaPipeline.start_link([])

```

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

### Start High-Throughput Producer with Partition Matching

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

Configure producer and processor concurrency to match or exceed the number of Kafka partitions for high-throughput scenarios. This example sets producer concurrency to 4 and processor concurrency to 12 for a topic with 12 partitions.

```elixir
Broadway.start_link(MyApp.HighThroughputPipeline,
  name: MyApp.HighThroughputPipeline,
  producer: [
    module: {BroadwayKafka.Producer, [
      hosts: [
        "broker1.kafka": 9092,
        "broker2.kafka": 9092,
        "broker3.kafka": 9092
      ],
      group_id: "high_throughput_group",
      topics: ["orders"],
      fetch_config: [
        max_bytes: 5_242_880,  # 5 MiB per fetch
        max_wait_time: 500
      ],
      offset_commit_on_ack: false,  # Commit on interval, not per-message
      group_config: [
        offset_commit_interval_seconds: 5
      ]
    ]},
    concurrency: 4   # 4 producers can hold up to 12 partitions total
  ],
  processors: [
    # At least as many processors as total expected partitions
    default: [concurrency: 12]
  ],
  batchers: [
    default: [
      concurrency: 12,      # One batcher lane per partition
      batch_size: 500,
      batch_timeout: 100
    ]
  ]
)
```

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

### BroadwayKafka.Producer Configuration

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

This example demonstrates how to configure and start a Broadway pipeline using BroadwayKafka.Producer. It includes required options like hosts, group_id, and topics, as well as optional settings for polling, offset management, group coordination, fetch tuning, and client configuration.

```APIDOC
## BroadwayKafka.Producer Configuration

### Description
Configure and start a Broadway Kafka pipeline using `BroadwayKafka.Producer`. This producer module handles the full lifecycle of Kafka consumption, including connection management, rebalancing, partition assignment, message fetching, and offset commits.

### Module
`BroadwayKafka.Producer`

### Required Options
- **hosts**: List of Kafka broker addresses (e.g., `[localhost:9092]`)
- **group_id**: The consumer group identifier.
- **topics**: List of Kafka topics to subscribe to.

### Optional Options

#### Polling/Offset Settings
- **receive_interval** (integer): Milliseconds between polls (default: 2000).
- **offset_commit_on_ack** (boolean): Commit offset on every acknowledgement (default: true).
- **offset_reset_policy** (:earliest | :latest | {:timestamp, ms}): Policy for resetting offsets.
- **begin_offset** (:assigned | :reset): Offset to begin consuming from.

#### Group Coordinator Configuration (`group_config`)
- **offset_commit_interval_seconds** (integer): Interval for committing offsets.
- **rejoin_delay_seconds** (integer): Delay before rejoining the consumer group.
- **session_timeout_seconds** (integer): Session timeout for the consumer group.
- **heartbeat_rate_seconds** (integer): Rate at which heartbeats are sent.
- **rebalance_timeout_seconds** (integer): Timeout for partition rebalancing.

#### Fetch Tuning (`fetch_config`)
- **min_bytes** (integer): Minimum bytes to fetch.
- **max_bytes** (integer): Maximum bytes to fetch (e.g., 1048576 for 1 MiB).
- **max_wait_time** (integer): Maximum time to wait for fetch.

#### Client Configuration (`client_config`)
- **ssl** (boolean): Enable SSL/TLS.
- **sasl** (tuple): SASL authentication configuration (e.g., `{:plain, "username", "password"}`).
- **connect_timeout** (integer): Connection timeout in milliseconds.
- **request_timeout** (integer): Request timeout in milliseconds.
- **client_id_prefix** (string): Prefix for the Kafka client ID.

#### Shared Client
- **shared_client** (boolean): Whether to share a single `brod` client across producers (default: false).

### Producer Concurrency
- **concurrency** (integer): Number of producer processes, which corresponds to the maximum parallel partition groups.

### Example Usage
```elixir
defmodule MyApp.KafkaPipeline do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__, 
      name: __MODULE__,
      producer: [
        module: {BroadwayKafka.Producer, [
          # Required
          hosts: [localhost: 9092],
          group_id: "my_consumer_group",
          topics: ["orders", "payments"],

          # Optional polling/offset settings
          receive_interval: 500,            
          offset_commit_on_ack: true,       
          offset_reset_policy: :earliest,   
          begin_offset: :assigned,          

          # Optional group coordinator config
          group_config: [
            offset_commit_interval_seconds: 5,
            rejoin_delay_seconds: 1,
            session_timeout_seconds: 30,
            heartbeat_rate_seconds: 5,
            rebalance_timeout_seconds: 30
          ],

          # Optional fetch tuning
          fetch_config: [
            min_bytes: 1,
            max_bytes: 1_048_576,   
            max_wait_time: 1_000
          ],

          # Optional client config
          client_config: [
            ssl: true,
            sasl: {:plain, "username", "password"},
            connect_timeout: 5_000,
            request_timeout: 30_000,
            client_id_prefix: "my_app-"
          ],

          shared_client: false
        ]},
        concurrency: 3   
      ],
      processors: [
        default: [concurrency: 10]
      ],
      batchers: [
        default: [
          batch_size: 100,
          batch_timeout: 200,
          concurrency: 5
        ]
      ]
    )
  end

  @impl Broadway
  def handle_message(_processor, message, _context) do
    message
  end

  @impl Broadway
  def handle_batch(_batcher, messages, batch_info, _context) do
    messages
  end
end

# Start the pipeline
{:ok, _pid} = MyApp.KafkaPipeline.start_link([])
```
```

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

### Start Shared Client Producer

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

Use `shared_client: true` for low-throughput pipelines with many topics under resource constraints. All producers share a single Kafka connection.

```elixir
Broadway.start_link(MyApp.SharedClientPipeline,
  name: MyApp.SharedClientPipeline,
  producer: [
    module: {BroadwayKafka.Producer, [
      hosts: [localhost: 9092],
      group_id: "shared_group",
      topics: ["topic_a", "topic_b", "topic_c"],
      shared_client: true,
      client_config: [
        client_id_prefix: "my_app_shared-"
      ]
    ]},
    # All 3 producers share one brod connection
    concurrency: 3
  ],
  processors: [default: [concurrency: 6]]
)
```

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

### Observe Kafka Assignment Revoke Events with Telemetry

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

Attach Telemetry handlers to monitor Kafka assignment revoke events, useful for measuring rebalance durations or triggering alerts. Events are emitted for start, stop, and exceptions during revocation.

```elixir
# Attach at application startup (e.g., in MyApp.Application.start/2)
:telemetry.attach_many(
  "broadway-kafka-handler",
  [
    [:broadway_kafka, :assignments_revoked, :start],
    [:broadway_kafka, :assignments_revoked, :stop],
    [:broadway_kafka, :assignments_revoked, :exception]
  ],
  &MyApp.TelemetryHandler.handle_event/4,
  nil
)

defmodule MyApp.TelemetryHandler do
  require Logger

  def handle_event([:broadway_kafka, :assignments_revoked, :start], _measurements, metadata, _cfg) do
    Logger.info("Kafka rebalance started for producer #{inspect(metadata.producer)}")
  end

  def handle_event([:broadway_kafka, :assignments_revoked, :stop], measurements, metadata, _cfg) do
    duration_ms = System.convert_time_unit(measurements.duration, :native, :millisecond)
    Logger.info("Kafka rebalance complete in #{duration_ms}ms for #{inspect(metadata.producer)}")
    MyApp.Metrics.histogram("kafka.rebalance.duration_ms", duration_ms)
  end

  def handle_event([:broadway_kafka, :assignments_revoked, :exception], _measurements, metadata, _cfg) do
    Logger.error("Kafka rebalance exception for #{inspect(metadata.producer)}")
  end
end
```

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

### Connect to Secured Kafka with SSL/TLS

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

Use the `:ssl` option within `:client_config` to enable SSL/TLS. Set it to `true` for system defaults or provide a keyword list of `tls_client_option` values for mutual TLS.

```elixir
Broadway.start_link(MyApp.SecurePipeline,
  name: MyApp.SecurePipeline,
  producer: [
    module: {BroadwayKafka.Producer, [
      hosts: ["kafka.example.com": 9093],
      group_id: "secure_group",
      topics: ["secure_events"],
      client_config: [
        ssl: [
          cacertfile: "/etc/ssl/certs/ca.pem",
          certfile:   "/etc/ssl/certs/client.pem",
          keyfile:    "/etc/ssl/private/client.key",
          verify:     :verify_peer
        ]
      ]
    ]},
    concurrency: 1
  ],
  processors: [default: [concurrency: 4]]
)
```

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

### Configure Broadway Producer with Kafka

Source: https://github.com/dashbitco/broadway_kafka/blob/main/README.md

Configure Broadway's producer to use `BroadwayKafka.Producer` by specifying Kafka hosts, group ID, and topics. Ensure the `concurrency` is set appropriately for your needs.

```elixir
defmodule MyBroadway do
  use Broadway

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

  def handle_message(_, message, _) do
    IO.inspect(message.data, label: "Got message")
    message
  end
end
```

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

### Add BroadwayKafka to Mix Dependencies

Source: https://github.com/dashbitco/broadway_kafka/blob/main/README.md

Add the :broadway_kafka dependency to your project's `mix.exs` file to include it in your application.

```elixir
def deps do
  [
    {:broadway_kafka, "~> 0.4.1"}
  ]
end
```

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

### Connect to Kafka with SASL Authentication

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

Configure SASL authentication using the `:sasl` option in `:client_config`. This supports plain credentials, credential files, or custom callback modules for advanced flows.

```elixir
# PLAIN with inline credentials
Broadway.start_link(MyApp.SaslPipeline,
  name: MyApp.SaslPipeline,
  producer: [
    module: {BroadwayKafka.Producer, [
      hosts: [localhost: 9092],
      group_id: "sasl_group",
      topics: ["secure_topic"],
      client_config: [
        # Inline credentials
        sasl: {:plain, "my_username", "my_password"},

        # Or SCRAM-SHA-256 with a credentials file path
        # sasl: {:scram_sha_256, "/run/secrets/kafka_credentials"},

        # Or a custom callback module
        # sasl: {:callback, MyApp.KafkaSaslCallback, []}
      ]
    ]},
    concurrency: 1
  ],
  processors: [default: [concurrency: 2]]
)
```

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

### Implement handle_failed/2 for Message Retries

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

Implement handle_failed/2 to manage messages that fail processing. This callback receives a list of failed messages and should forward them to a dead-letter queue or another system.

```elixir
defmodule MyApp.ResilientPipeline do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {BroadwayKafka.Producer, [
          hosts: [localhost: 9092],
          group_id: "resilient_group",
          topics: ["events"]
        ]},
        concurrency: 1
      ],
      processors: [default: [concurrency: 4]]
    )
  end

  @impl Broadway
  def handle_message(_processor, message, _context) do
    case process(message.data) do
      :ok    -> message
      :error -> Broadway.Message.failed(message, "processing_error")
    end
  end

  # Called with all messages that were marked as failed via
  # Broadway.Message.failed/2 anywhere in the pipeline.
  @impl Broadway
  def handle_failed(messages, _context) do
    Enum.each(messages, fn message ->
      %{topic: topic, partition: partition, offset: offset} = message.metadata

      IO.warn("Failed: #{topic}/#{partition}@#{offset}")

      # Forward to a dead-letter Kafka topic via :brod
      :brod.produce_sync(
        :my_brod_client,
        "events_dlq",
        _partition = 0,
        _key = "",
        message.data
      )
    end)

    # Must return the list (Broadway will still ack them all)
    messages
  end

  defp process(_data), do: :ok
end
```

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

### Control Kafka Consumer Offset Reset Policy

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

Use `:offset_reset_policy` to manage consumer group offsets on initial startup or when offsets expire. Combine with `:begin_offset` to control resumption or resetting.

```elixir
# Consume from the very beginning of all retained messages
Broadway.start_link(MyApp.ReplayPipeline,
  name: MyApp.ReplayPipeline,
  producer: [
    module: {BroadwayKafka.Producer, [
      hosts: [localhost: 9092],
      group_id: "replay_group",
      topics: ["events"],
      # Reset to earliest when no committed offset exists
      offset_reset_policy: :earliest,
      # Force offset reset on start (ignore stored group offsets)
      begin_offset: :reset,
    ]},
    concurrency: 1
  ],
  processors: [default: [concurrency: 4]]
)
```

```elixir
# Consume only messages produced after a specific Unix timestamp (ms)
Broadway.start_link(MyApp.TimestampPipeline,
  name: MyApp.TimestampPipeline,
  producer: [
    module: {BroadwayKafka.Producer, [
      hosts: [localhost: 9092],
      group_id: "timestamp_group",
      topics: ["events"],
      # Seek to the first message at or after 2024-01-01T00:00:00Z
      offset_reset_policy: {:timestamp, 1_704_067_200_000},
      begin_offset: :reset
    ]},
    concurrency: 1
  ],
  processors: [default: [concurrency: 4]]
)
```

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

### Dynamically Update Subscribed Topics with update_topics/2

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

Use BroadwayKafka.update_topics/2 to dynamically change the topics a pipeline subscribes to at runtime. This function replaces the entire topic list, so ensure all desired topics are included in the new list.

```elixir
# Initial pipeline subscribes to ["topic_a"]
{:ok, _pid} = Broadway.start_link(MyApp.KafkaPipeline, 
  name: MyApp.KafkaPipeline,
  producer: [
    module: {BroadwayKafka.Producer, [
      hosts: [localhost: 9092],
      group_id: "dynamic_group",
      topics: ["topic_a"]
    ]},
    concurrency: 1
  ],
  processors: [default: [concurrency: 2]]
)

# Later — add "topic_b" and "topic_c" dynamically
:ok = BroadwayKafka.update_topics(MyApp.KafkaPipeline, ["topic_a", "topic_b", "topic_c"])

# Remove all topics (pause consumption)
:ok = BroadwayKafka.update_topics(MyApp.KafkaPipeline, [])
```

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

### Access Kafka Message Metadata in handle_message/3

Source: https://context7.com/dashbitco/broadway_kafka/llms.txt

Access Kafka record fields directly from the message's :metadata map within the handle_message/3 callback. This allows routing based on topic or decoding the payload.

```elixir
defmodule MyApp.OrderPipeline do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {BroadwayKafka.Producer, [
          hosts: [localhost: 9092],
          group_id: "order_group",
          topics: ["orders"]
        ]},
        concurrency: 1
      ],
      processors: [default: [concurrency: 5]]
    )
  end

  @impl Broadway
  def handle_message(_processor, message, _context) do
    %{ 
      topic:     topic,
      partition: partition,
      offset:    offset,
      key:       key,
      ts:        timestamp,
      headers:   headers
    } = message.metadata

    # Route messages to different batchers based on metadata
    message = 
      case topic do
        "orders" -> Broadway.Message.put_batcher(message, :orders)
        _other   -> Broadway.Message.put_batcher(message, :default)
      end

    # Decode JSON payload
    case Jason.decode(message.data) do
      {:ok, payload} ->
        IO.puts("Order #{payload["id"]} at offset #{offset} on #{topic}/#{partition}")
        Broadway.Message.update_data(message, fn _ -> payload end)

      {:error, reason} ->
        IO.warn("Bad JSON at offset #{offset}: #{inspect(reason)}")
        Broadway.Message.failed(message, "invalid_json")
    end
  end
end
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.