### 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 ``` ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.