KafkaBatcher

A library to increase the throughput of producing messages (coming one at a time) to Kafka by accumulating these messages into batches.

Installation

1. Add KafkaBatcher to your list of dependencies in mix.exs:

  def deps do
    [
      {:kafka_batcher, "~> 1.0.0"},
      # and one of kafka libraries
      # {:kaffe, "~> 1.24"}
      # or
      # {:kafka_ex, "~> 0.12"}
    ]
  end

1. Add KafkaBatcher supervisor to your supervisor tree

  def start(_type, _args) do
    children = [
      # Describe the child spec
      KafkaBatcher.Supervisor
    ]

    opts = [strategy: :one_for_one, name: MyApp.Supervisor, max_restarts: 3, max_seconds: 5]
    Supervisor.start_link(children, opts)
  end

1. Configure a KafkaBatcher Producer

Config example:

config :kafka_batcher, KafkaBatcher.Collector1, topic_name: "topic1"
config :kafka_batcher, KafkaBatcher.Collector2, topic_name: "topic2"
config :kafka_batcher, KafkaBatcher.Collector3, topic_name: "topic3"
config :kafka_batcher, KafkaBatcher.Collector4, topic_name: "topic4"
config :kafka_batcher, KafkaBatcher.Collector5, topic_name: "topic5"

config :kafka_batcher, collectors:
        [
          KafkaBatcher.Collector1,
          KafkaBatcher.Collector2,
          KafkaBatcher.Collector3,
          KafkaBatcher.Collector4,
          KafkaBatcher.Collector5
        ]
        
config :kafka_batcher, :kafka,
  endpoints: "localhost:9092",
  # in case you use SASL
  # sasl: %{mechanism: :scram_sha_512, login: "login", password: "password"},
  # ssl: true,
  telemetry: true,
  allow_topic_auto_creation: false,
  kafka_topic_aliases: %{
    "real_topic_name1" => "incoming-events",
    "real_topic_name2" => "special-topic"
  }

# In case you use KafkaEx, you need to disable default worker to avoid crashes
config :kafka_ex, :disable_default_worker, true

Available parameters:

• :required_acks How many acknowledgements the kafka broker should receive from the clustered replicas before acking producer.
• :endpoints Kafka cluster endpoints, can be any of the brokers in the cluster, which does not necessarily have to be the leader of any partition, e.g. a load-balanced entrypoint to the remote Kafka cluster. More information in the Brod producer docs.
• :telemetry, if set to true, metrics will be collected and exposed with PromEx.
• :allow_topic_auto_creation, if set true, topics automatically are created with default parameters.
• :partition_strategy, allows values: :random, :md5 or function (e.g. fn _topic, _partitions_count, key, _value -> key end)
• :partition_fn, a function that takes four arguments and returns a number of topic (see below)
• :collect_by_partition, if set to true, producer accumulates messages separately for each partition of the topic
• :batch_size, count of messages to be accumulated by collector before producing
• :max_wait_time, max interval between batches in milliseconds. The batch will be produced to Kafka either by batch_size or by max_wait_time parameter.
• :batch_flusher, a module implementing a function flush?(binary(), map()) :: boolean(). If the function returns true, the current batch will be sent to Kafka immediately.
• :kafka_topic_aliases - you could define custom aliases for Kafka topics, it doesn't affect anything except metric labels.
• :sasl - optional parameter. The parameter includes three parameters: %{mechanism: mechanism, login: "login", password: "password"}. Mechanism is plain | scram_sha_256 | scram_sha_512. Login and password should be type binary().
• :ssl - optional parameter. Ssl should be type boolean(). By default :ssl is false.
• :min_delay - optional parameter. Set minimal delay before send events. This parameter allows to increase max throughput in case when you get more messages (in term of count per second) than you expected when set batch_size parameter.
• :max_batch_bytesize - optional parameter. Allows to set a limit on the maximum batch size. By default it is 1_000_000 bytes.

Important: The size of one message should not exceed max_batch_bytesize setting. If you need to work with large messages you must increase max_batch_bytesize value and value of Kafka topic setting max.message.bytes as well.

Note: you can still produce messages to any Kafka topic (even if it is not described in the kafka_batcher config) by using direct calls of Kaffe or KafkaEx.

Usage

Collector examples

  defmodule MyApp.Collector1 do
    use KafkaBatcher.Collector, 
      collect_by_partition: true,
      partition_fn: &MyApp.Collector1.calculate_partition/4,
      required_acks: -1,
      batch_size: 30,
      max_wait_time: 20_000
                                        
    def calculate_partition(_topic, partitions_count, _key, value) do
      val = value["client_id"] || value["device_id"]
      :erlang.phash2(val, partitions_count)
    end
  end
  
  defmodule MyApp.Collector2 do
    use KafkaBatcher.Collector, 
      collect_by_partition: false,
      required_acks: 0,
      batch_size: 10,
      max_wait_time: 20_000
    end

  defmodule MyApp.Collector3 do
    use KafkaBatcher.Collector, 
      collect_by_partition: false,
      required_acks: 0,
      batch_size: 10,
      max_wait_time: 20_000,
      partition_strategy: :random
  end

  defmodule MyApp.Collector4 do
    use KafkaBatcher.Collector, 
      collect_by_partition: true,
      required_acks: 0,
      partition_fn: &MyApp.Collector4.calculate_partition/4,
      batch_size: 50,
      batch_flusher: MyApp.Collector4.BatchFlusher

    def calculate_partition(_topic, partitions_count, _key, value) do
      rem(key, partitions_count)
    end
    
    defmodule BatchFlusher do
      def flush?(_key, %{"type" => "SpecialType"}) do
        true
      end

      def flush?(_key, _value) do
        false
      end
    end
  end
  
  defmodule MyApp.Collector5 do
    use KafkaBatcher.Collector, 
      collect_by_partition: false,
      partition_strategy: :md5,
      batch_size: 100,
      max_wait_time: 20_000,
      min_delay: 100,
      max_batch_bytesize: 1_000_000
  end

Collector usage

  defmodule MyApp.MyModule do
    ...
    def produce_to_kafka_topic1(event)  do
      MyApp.Collector1.add_events(event)
    end

    def produce_to_kafka_topic2(event)  do
      MyApp.Collector2.add_events(event)
    end
    ...
  end

Getting current config of topic

  KafkaBatcher.Config.get_collector_config("topic1")

Getting all topics with config

  KafkaBatcher.Config.get_configs_by_topic()

Testing

mix test

or

mix test --cover

see https://github.com/parroty/excoveralls for details

Prometheus metrics

The library exposes the following metrics using the PromEx exporter plugin:

• prom_ex_kafka_consumer_batch_messages_count (histogram) - The count of messages in one batch consumer.
• prom_ex_kafka_consumer_batch_total_size_byte (histogram) - The size of a batch consumer of messages.
• prom_ex_kafka_consumer_duration_seconds (histogram) - The time to produce one batch to Kafka.
• prom_ex_kafka_producer_batch_messages_count (histogram) - The count of messages in one batch producer.
• prom_ex_kafka_producer_batch_total_size_byte (histogram) - The size of a batch producer of messages.
• prom_ex_kafka_producer_duration_seconds (histogram) - The time to produce one batch to Kafka.