Kaffe.Producer (Kaffe v1.27.2)

View Source

The producer pulls in values from the Kaffe producer configuration:

  • heroku_kafka_env - endpoints and SSL configuration will be pulled from ENV
  • endpoints - plaintext Kafka endpoints
  • topics - a list of Kafka topics to prep for producing
  • partition_strategy - the strategy to use when selecting the next partition. Default :md5.
    • :md5: provides even and deterministic distrbution of the messages over the available partitions based on an MD5 hash of the key
    • :random - Select a random partition
    • function - Pass a function as an argument that accepts five arguments and returns the partition number to use for the message
      • topic, current_partition, partitions_count, key, value

Clients can also specify a partition directly when producing.

Currently only synchronous production is supported.

Summary

Types

headers()

Headers represent a list of tuples containing key, value pairs of type binary()

message()

A Kafka message can be represented as a tuple containing a key value pair of type binary()

message_object()

A Kafka message can also be represented as a map, containing :key, :value, and :headers

Functions

produce(topic, message_list, opts \\ [])

Synchronously produce the messages_list to topic

produce_sync(topic, message_list)

Synchronously produce the given key/value to the first Kafka topic.

produce_sync(topic, partition, message_list)

Synchronously produce the key/value to topic

produce_sync(topic, partition, key, value)

Synchronously produce the given key/value to the topic/partition

start_producer_client()

Types

headers()

@type headers() :: [{key :: binary(), value :: binary()}]

Headers represent a list of tuples containing key, value pairs of type binary()

message()

@type message() :: {key :: binary(), value :: binary()}

A Kafka message can be represented as a tuple containing a key value pair of type binary()

message_object()

@type message_object() :: %{
  key: key :: binary(),
  value: value :: binary(),
  headers: headers :: headers()
}

A Kafka message can also be represented as a map, containing :key, :value, and :headers

Functions

produce(topic, message_list, opts \\ [])

Synchronously produce the messages_list to topic

  • messages_list must be a list of type message() or message_object()
  • opts may include the partition strategy to use, partition_strategy: :md5, or :random or a function.

Returns:

 * `:ok` on successfully producing each message
 * `{:error, reason}` for any error

produce_sync(topic, message_list)

Synchronously produce the given key/value to the first Kafka topic.

This is a simpler way to produce if you've only given Producer a single topic for production and don't want to specify the topic for each call.

Returns:

 * `:ok` on successfully producing the message
 * `{:error, reason}` for any error

produce_sync(topic, partition, message_list)

Synchronously produce the key/value to topic

See produce_sync/2 for returns.

produce_sync(topic, partition, key, value)

Synchronously produce the given key/value to the topic/partition

See produce_sync/2 for returns.

start_producer_client()