Defines a kafka client.

To use it you must do 3 steps:

• Use it in a module
• Config the module on your config file
• Start the module on your supervision tree

Using it in a module

When used it expects an :otp_app option that is the OTP application that has the client configuration.

defmodule MyApp.MyClient do
  use Klife.Client, otp_app: :my_app
end

Configuration

The client has a bunch of configuration options, you can read more below. But it will look somehting like this:

config :my_app, MyApp.MyClient,
  connection: [
    bootstrap_servers: ["localhost:19092", "localhost:29092"],
    ssl: false
  ],
  producers: [
    [
      name: :my_custom_producer,
      linger_ms: 5,
      max_in_flight_requests: 10
    ]
  ],
  topics: [
    [name: "my_topic_0", producer: :my_custom_producer]
  ]

You can see more configuration examples on the "Client configuration examples" section or an working application example on the example folder on the project's repository.

Configuration options:

• :connection (non-empty keyword/0) - Required.

  • :bootstrap_servers (list of String.t/0) - Required. List of servers to establish the initial connection. (eg: ["localhost:9092", "localhost:9093"])

  • :ssl (boolean/0) - Specify the underlying socket module. Use :ssl if true and :gen_tcp if false. The default value is false.

  • :connection_count (pos_integer/0) - How many TCP connections the client will maintain for each broker. The default value is 1.

  • :connect_opts (list of term/0) - Options used to configure the socket connection, which are forwarded to the connect/3 function of the underlying socket module (see ssl option above.). The default value is [inet_backend: :socket, active: false].

  • :socket_opts (list of term/0) - Options used to configure the open socket, which are forwarded to the setopts/2 function of the underlying socket module :inet for :gen_tcp and :ssl for :ssl (see ssl option above.). The default value is [keepalive: true].

  • :sasl_opts (list of term/0) - Options to configure SASL authentication, see SASL section for supported mechanisms and examples. The default value is [].

  • :default_request_timeout_ms (pos_integer/0) - The default request timeout ms to be used on all requests to brokers. Some other options may override this value in specific scenarios The default value is 30000.

• :default_producer (atom/0) - Name of the producer to be used on produce API calls when a specific producer is not provided via configuration or option. If not provided a default producer will be started automatically. The default value is :klife_default_producer.

• :producers (List of Klife.Producer configurations) - List of configurations, each starting a new producer for use with produce api. The default value is [].

• :default_partitioner (atom/0) - Partitioner module to be used on produce API calls when a specific partitioner is not provided via configuration or option. The default value is Klife.Producer.DefaultPartitioner.

• :default_txn_pool (atom/0) - Name of the txn pool to be used on transactions when a :pool_name is not provided as an option. If not provided a default txn pool will be started automatically. The default value is :klife_default_txn_pool.

• :txn_pools (List of Klife.TxnProducerPool configurations) - List of configurations, each starting a pool of transactional producers for use with transactional api. The default value is [].

• :topics (List of Klife.Topic configurations) - List of topics that may have special configurations The default value is [].

• :default_fetcher (atom/0) - Name of the default fetcher to be used on the consumer API when a specific fetcher is not provided via configuration or option. If not provided a default fetcher will be started automatically. The default value is :klife_default_fetcher.

• :fetchers (List of Klife.Consumer.Fetcher configurations) - List of configurations, each starting a new fetcher to be used with consumer api. The default value is [].

• :disabled_features (List atoms representing a features to disable.) - :producer disable producer feature. :txn_producer disables transactions. The default value is [].

• :enable_unkown_topics (boolean/0) - Define if the client will be able to work with non configured topics. When true the client will collect metadata for all known topics of the cluster, this may have performance impacts on large clusters. When set to false the client will collect metadata only for topics defined on the topics options. The default value is true.

Starting it

Finally, it must be started on your application. It will look something like this:

defmodule MyApp.Application do
  def start(_type, _args) do
    children = [
      # some other modules...,
      MyApp.MyClient
    ]

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

Producer API overview

In order to interact with the producer API you will work with Klife.Record module as your main input and output data structure.

Usually you will give an record to some producer API function and it will return an enriched record with some new attributes based on what happened.

So in summary the interaction goes like this:

• Build one or more Klife.Record
• Pass it to some producer API function
• Receive an enriched version of the provided records

rec = %Klife.Record{value: "some_val", topic: "my_topic_1"}
{:ok, %Klife.Record{offset: offset, partition: partition}} = MyClient.produce(rec)

Consumer API overview

For continuous consumption of records, the recommended approach is to use Klife.Consumer.ConsumerGroup. It handles partition assignment, offset tracking, rebalancing, and fault tolerance automatically. See the Klife.Consumer.ConsumerGroup docs for details.

The client also exposes lower-level fetch functions fetch/4, fetch/2, and fetch_async/4, for cases where a full consumer group is not needed. These are intended for standalone use: one-off reads, backfilling, debugging, or any workflow where you manage offsets yourself.

The fetch functions return Klife.Record structs, the same data structure used throughout the producer API. Each returned record is fully enriched with all the relevant data available.

# Fetch a batch of records starting at a given offset
{:ok, records} = MyClient.fetch("my_topic_1", 0, 42)

# Fetch from multiple topic/partition/offset combinations in one request
tpo_list = [{"my_topic_1", 0, 42}, {"my_topic_2", 3, 0}]
%{{"my_topic_1", 0, 42} => {:ok, _}} = MyClient.fetch(tpo_list)

# Async fetch result delivered as a message to the calling process
{:ok, _pid} = MyClient.fetch_async("my_topic_1", 0, 42)