repository·main·Indexed 23 days ago
https://github.com/zendesk/ruby-kafkaA Ruby client library for Apache Kafka designed for operational simplicity, providing logging and metrics for debugging. It supports synchronous and asynchronous production, consumer groups, and various partitioning strategies. Note: This library is no longer actively developed and has been superseded by rdkafka-ruby; it is not recommended for production usage.
ruby-kafka has been superseded by librdkafka via rdkafka-ruby bindings. While it may receive security patches and bug fixes, it is not recommended for production usage. For modern Kafka features and official support, use rdkafka-ruby instead.
Kafka topics are divided into partitions. You can control how messages are assigned to these partitions using several strategies:
Distributes messages evenly across partitions to ensure balanced consumer load. If no key is provided, the producer assigns a partition randomly.
Assigns messages to partitions based on a specific property (e.g., a session_id). This ensures that all messages related to a specific entity are processed by the same consumer instance, facilitating local state management.
To ensure compatibility with other Kafka clients (like Java-based ones), you can implement a custom partitioner. You can pass a class that responds to call(partition_count, message) or a Proc to the Kafka.new constructor.
The Kafka::Consumer API is designed for flexibility and stability using a loop-based approach rather than a rigid object model.
Key features include:
ruby-kafka implements at-least-once delivery guarantees for the synchronous producer. This means a message is guaranteed to be delivered, but it may be delivered more than once if a failure occurs during acknowledgement.
For the synchronous producer, once #deliver_messages returns, you can be certain the message has been received by Kafka.
Caveats:
required_acks is set to 0, there is no guarantee of delivery.#deliver_messages does not guarantee delivery has completed.You can tune three parameters to balance the number of messages processed (throughput) against the time it takes to process them (latency):
min_bytes: Minimum bytes to return from a single fetch. Higher values increase throughput. (Default: 1 byte).max_wait_time: Maximum seconds to wait before returning data. High values increase throughput; low values decrease latency. This overrides min_bytes. (Default: 1 second).max_bytes_per_partition: Maximum data a broker returns for a single partition. Increasing this improves throughput. (Default: 1MB).min_bytes and max_wait_time are passed to #each_message or #each_batch. max_bytes_per_partition is passed to #subscribe.
# Waits for data for up to 5 seconds on each broker, preferring to fetch at least 5KB at a time.
consumer.each_message(min_bytes: 1024 * 5, max_wait_time: 5) do |message|
# ...
end
# Fetches up to 5MB per partition at a time for better throughput.
consumer.subscribe("greetings", max_bytes_per_partition: 5 * 1024 * 1024)The Consumer API uses Kafka's Consumer Groups feature to allow multiple consumer processes to coordinate access to a topic.
Key features:
group_id.require "kafka"
kafka = Kafka.new(["kafka1:9092", "kafka2:9092"])
# Consumers with the same group id will form a Consumer Group together.
consumer = kafka.consumer(group_id: "my-consumer")
# It's possible to subscribe to multiple topics by calling `subscribe`
# repeatedly.
consumer.subscribe("greetings")
# Stop the consumer when the SIGTERM signal is sent to the process.
# It's better to shut down gracefully than to kill it.
trap("TERM") { consumer.stop }
# This will loop indefinitely, yielding each message in turn.
consumer.each_message do |message|
puts message.topic, message.partition
puts message.offset, message.key, message.value
endThe library is agnostic to serialization formats. It treats both the message value and the key as binary strings. You are responsible for encoding your data (e.g., to JSON or Avro) before passing it to the producer.
require "json"
event = {
"name" => "pageview",
"url" => "https://example.com/posts/123"
}
data = JSON.dump(event)
producer.produce(data, topic: "events")The Kafka::Producer is designed for resilience and observability. It uses two internal data structures: a pending messages list and a message buffer.
#produce: When you call Kafka::Producer#produce, the message is added to the pending list. No network communication occurs at this stage, so network errors are not raised immediately.#deliver_messages: This method triggers the actual network activity. It:produce requests to brokers.Error Handling: If messages remain in the pending list or the buffer after #deliver_messages completes, a Kafka::DeliveryFailed exception is raised. You must rescue this exception and decide how to retry (e.g., by calling #deliver_messages again later).
The library is organized into several layers. As an end-user, you should primarily interact with the API layer and the configuration layer. The other layers are considered internal and may change without warning.
Kafka::Client implements the public APIs, and Kafka.new is a convenience method to instantiate it.Kafka::Consumer for consuming messages, and Kafka::Producer or Kafka::AsyncProducer for producing messages.Kafka::Cluster or Kafka::Broker).The Kafka::AsyncProducer provides a hands-off approach compared to the synchronous producer, trading fine-grained control for ease of use and resilience.
When using ruby-kafka in multi-threaded environments, follow these concurrency rules:
Kafka client object between threads. Network communication is not synchronized.#each_message or #each_batch block. Returning from the block triggers checkpointing, which assumes the yielded messages have been successfully processed.You can install ruby-kafka using Bundler by adding it to your Gemfile, or by installing the gem directly via the command line.
Add this line to your Gemfile:
gem 'ruby-kafka'Then run:
bundlegem install ruby-kafka