Idiomatic Elixir interface to the Kafka protocol.
Kayrock provides serialization and deserialization for all Kafka protocol messages. It generates Elixir structs from the Kafka protocol schema, allowing you to work with Kafka at the wire protocol level.
Core Features
- Generated structs for every Kafka API version
- Serialization to wire protocol (iodata)
- Deserialization from binary responses
- Compression support (gzip, snappy, LZ4, Zstandard)
Summary
Types
Represents an API response message
Specifies the version of an API message
A pid for a Kayrock.BrokerConnection process.
A pid for a Kayrock.Client process.
A broker's advertised integer node id
A node selector for sending messages to the cluster
Functions
Fetch the supported API versions from the cluster Kayrock currently supports versions 0 and 1. ApiVersions
Makes a synchronous call directly to a broker
Make an api call to a Kafka cluster mediated through a client
Create one or more topics
Delete topics from the Kafka cluster.
Fetch messages from a Kafka topic.
Produce messages to a Kafka topic.
Fetch metadata for topics from the Kafka cluster.
Types
@type api_response() :: map()
Represents an API response message
This will generally be a generated struct.
@type api_version() :: non_neg_integer()
Specifies the version of an API message
Kayrock will generally try to pick a reasonable default here, but you can override it in many cases. See Kafka Protocol Documentation
@type broker_pid() :: pid()
A pid for a Kayrock.BrokerConnection process.
This is for low-level communication with individual brokers. Generally you
should work with a client_pid.
@type client_pid() :: pid()
A pid for a Kayrock.Client process.
@type node_id() :: integer()
A broker's advertised integer node id
@type node_selector() :: node_id() | :random | :controller | {:topic_partition, topic_name(), partition_id()}
A node selector for sending messages to the cluster
Generally the Kayrock API will select the right node depending on the operation. These are exposed to provide fine-grained user control in a few cases where it might make sense to override the default.
Possible values:
- Integer - Directly access a node by id. Generally you should not do this unless you are operating at a low level.
:random- Select a node at random from the cluster metadata.:controller- Select the controller node - this is used for cluster management messages
@type partition_id() :: non_neg_integer()
@type topic_name() :: binary()
Functions
@spec api_versions(client_pid(), api_version(), node_selector()) :: {:ok, api_response()}
Fetch the supported API versions from the cluster Kayrock currently supports versions 0 and 1. ApiVersions
@spec broker_sync_call(pid(), Kayrock.Request.t()) :: {:ok, map()} | {:error, term()}
Makes a synchronous call directly to a broker
broker_id should be the pid of a Kayrock.BrokerConnection. request must
have an implementation for the Kayrock.Request protocol.
@spec client_call(client_pid(), Kayrock.Request.t(), node_selector()) :: {:ok, api_response()}
Make an api call to a Kafka cluster mediated through a client
Create one or more topics
Delete topics from the Kafka cluster.
Fetch messages from a Kafka topic.
Options
:version- Protocol version to use (default: 4):max_wait_time- Maximum wait time in ms (default: 1000):min_bytes- Minimum bytes to return (default: 0):max_bytes- Maximum bytes to return (default: 1_000_000):isolation_level- 0 = read uncommitted, 1 = read committed (default: 1)
Produce messages to a Kafka topic.
Options
:version- Protocol version to use (default: 1):acks- Acknowledgment level (default: -1 for all replicas):timeout- Request timeout in ms (default: 1000)
Fetch metadata for topics from the Kafka cluster.
Parameters
client_pid- PID of aKayrock.Clientprocesstopics- List of topic names, ornilfor all topics