DOCSEARCHbyAlgolia
Submit Docs

BroadwayKafka

repository·main·Indexed 42 days ago

https://github.com/elixir-broadway/broadway_kafka

A Kafka connector for the Broadway processing framework that allows developers to use Kafka as a data source within a Broadway pipeline. Includes the BroadwayKafka.Producer module for configuration and BroadwayKafka.update_topics/2 for dynamically updating consumed topics in running pipelines.

Tokens
1K
Snippets
4
Records
4
Agent score
4%

What's inside BroadwayKafka

  1. Configure BroadwayKafka.Producer

    main

    To use Kafka as a source for your Broadway pipeline, configure the producer option in Broadway.start_link/2 using the BroadwayKafka.Producer module.

    Required configuration options for the producer include:

    • hosts: A list of Kafka brokers (e.g., [localhost: 9092]).
    • group_id: A string identifying the consumer group.
    • topics: A list of Kafka topics to consume from.
    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
    )
  end

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

  3. Set up a local Kafka and Zookeeper environment with Docker Compose

    main

    You can use the provided docker-compose.yml to spin up a local Kafka broker and Zookeeper instance for development and testing. This setup configures Kafka with two listeners: an INTERNAL listener for communication within the Docker network and an EXTERNAL listener accessible via localhost:9092 from your host machine.

    version: '3.9'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "127.0.0.1:2181:2181"
  kafka:
    image: wurstmeister/kafka:2.13-2.7.1
    ports:
      - "127.0.0.1:9092:9092"
    environment:
      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9092"
      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka:29092,EXTERNAL://localhost:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    depends_on:
      - zookeeper

  4. Update Kafka topics in a running Broadway producer with update_topics/2

    main

    Use BroadwayKafka.update_topics/2 to sequentially update the list of topics being consumed by all Broadway producers within a specific Broadway pipeline. This is useful for dynamically changing the set of Kafka topics a running pipeline is subscribed to without restarting the entire application.

    Parameters:

    • name: The name of the Broadway pipeline (e.g., MyBroadway).
    • topics: An enumerable of topic names (strings) that should be active in the producers.
    # Update the pipeline to consume only 'topic_a' and 'topic_b'
BroadwayKafka.update_topics(MyBroadway, ["topic_a", "topic_b"])

# Clear all topics (empty list)
BroadwayKafka.update_topics(MyBroadway, [])