ruby-kafka

Deprecation notice

This library is no longer actively developed and has been superseded by librdkafka via rdkafka-ruby bindings. While this library may still receive security patches and bug fixes, it is no longer recommended for production usage.

There needs to be a concerted effort to keep up with Kafka features. There is no point in trying to keep up with Kafka development when other languages use a well-established C binding and get more official support.

ruby-kafka

A Ruby client library for Apache Kafka, a distributed log and message bus. The focus of this library will be operational simplicity, with good logging and metrics that can make debugging issues easier.

Table of Contents

1. Installation
2. Compatibility
3. Usage
   1. Setting up the Kafka Client
   2. Producing Messages to Kafka
      1. Efficiently Producing Messages
      2. Asynchronously Producing Messages
      3. Serialization
      4. Partitioning
      5. Buffering and Error Handling
      6. Message Durability
      7. Message Delivery Guarantees
      8. Compression
      9. Producing Messages from a Rails Application
   3. Consuming Messages from Kafka
      1. Consumer Groups
      2. Consumer Checkpointing
      3. Topic Subscriptions
      4. Shutting Down a Consumer
      5. Consuming Messages in Batches
      6. Balancing Throughput and Latency
      7. Customizing Partition Assignment Strategy
   4. Thread Safety
   5. Logging
   6. Instrumentation
   7. Monitoring
      1. What to Monitor
      2. Reporting Metrics to Statsd
      3. Reporting Metrics to Datadog
   8. Understanding Timeouts
   9. Security
      1. Encryption and Authentication using SSL
      2. Authentication using SASL
   10. Topic management
4. Design
   1. Producer Design
   2. Asynchronous Producer Design
   3. Consumer Design
5. Development
6. Support and Discussion
7. Roadmap
8. Higher level libraries
   1. Message processing frameworks
   2. Message publishing libraries

Installation

Add this line to your application's Gemfile:

gem 'ruby-kafka'

And then execute:

$ bundle

Or install it yourself as:

$ gem install ruby-kafka

Compatibility

Producer API
Consumer API




Kafka 0.8
Full support in v0.4.x
Unsupported




Kafka 0.9
Full support in v0.4.x
Full support in v0.4.x




Kafka 0.10
Full support in v0.5.x
Full support in v0.5.x




Kafka 0.11
Full support in v0.7.x
Limited support




Kafka 1.0
Limited support
Limited support




Kafka 2.0
Limited support
Limited support




Kafka 2.1
Limited support
Limited support




Kafka 2.2
Limited support
Limited support




Kafka 2.3
Limited support
Limited support




Kafka 2.4
Limited support
Limited support




Kafka 2.5
Limited support
Limited support




Kafka 2.6
Limited support
Limited support




Kafka 2.7
Limited support
Limited support

This library is targeting Kafka 0.9 with the v0.4.x series and Kafka 0.10 with the v0.5.x series. There's limited support for Kafka 0.8, and things should work with Kafka 0.11, although there may be performance issues due to changes in the protocol.

• Kafka 0.8: Full support for the Producer API in ruby-kafka v0.4.x, but no support for consumer groups. Simple message fetching works.
• Kafka 0.9: Full support for the Producer and Consumer API in ruby-kafka v0.4.x.
• Kafka 0.10: Full support for the Producer and Consumer API in ruby-kafka v0.5.x. Note that you must run version 0.10.1 or higher of Kafka due to limitations in 0.10.0.
• Kafka 0.11: Full support for Producer API, limited support for Consumer API in ruby-kafka v0.7.x. New features in 0.11.x includes new Record Batch format, idempotent and transactional production. The missing feature is dirty reading of Consumer API.
• Kafka 1.0: Everything that works with Kafka 0.11 should still work, but so far no features specific to Kafka 1.0 have been added.
• Kafka 2.0: Everything that works with Kafka 1.0 should still work, but so far no features specific to Kafka 2.0 have been added.
• Kafka 2.1: Everything that works with Kafka 2.0 should still work, but so far no features specific to Kafka 2.1 have been added.
• Kafka 2.2: Everything that works with Kafka 2.1 should still work, but so far no features specific to Kafka 2.2 have been added.
• Kafka 2.3: Everything that works with Kafka 2.2 should still work, but so far no features specific to Kafka 2.3 have been added.
• Kafka 2.4: Everything that works with Kafka 2.3 should still work, but so far no features specific to Kafka 2.4 have been added.
• Kafka 2.5: Everything that works with Kafka 2.4 should still work, but so far no features specific to Kafka 2.5 have been added.
• Kafka 2.6: Everything that works with Kafka 2.5 should still work, but so far no features specific to Kafka 2.6 have been added.
• Kafka 2.7: Everything that works with Kafka 2.6 should still work, but so far no features specific to Kafka 2.7 have been added.

This library requires Ruby 2.1 or higher.

Usage

Please see the documentation site for detailed documentation on the latest release. Note that the documentation on GitHub may not match the version of the library you're using – there are still being made many changes to the API.

Setting up the Kafka Client

A client must be initialized with at least one Kafka broker, from which the entire Kafka cluster will be discovered. Each client keeps a separate pool of broker connections. Don't use the same client from more than one thread.

require "kafka"

# The first argument is a list of "seed brokers" that will be queried for the full
# cluster topology. At least one of these *must* be available. `client_id` is
# used to identify this client in logs and metrics. It's optional but recommended.
kafka = Kafka.new(["kafka1:9092", "kafka2:9092"], client_id: "my-application")

You can also use a hostname with seed brokers' IP addresses:

kafka = Kafka.new("seed-brokers:9092", client_id: "my-application", resolve_seed_brokers: true)

Producing Messages to Kafka

The simplest way to write a message to a Kafka topic is to call #deliver_message:

kafka = Kafka.new(...)
kafka.deliver_message("Hello, World!", topic: "greetings")

This will write the message to a random partition in the greetings topic. If you want to write to a specific partition, pass the partition parameter:

# Will write to partition 42.
kafka.deliver_message("Hello, World!", topic: "greetings", partition: 42)

If you don't know exactly how many partitions are in the topic, or if you'd rather have some level of indirection, you can pass in partition_key instead. Two messages with the same partition key will always be assigned to the same partition. This is useful if you want to make sure all messages with a given attribute are always written to the same partition, e.g. all purchase events for a given customer id.

# Partition keys assign a partition deterministically.
kafka.deliver_message("Hello, World!", topic: "greetings", partition_key: "hello")

Kafka also supports message keys. When passed, a message key can be used instead of a partition key. The message key is written alongside the message value and can be read by consumers. Message keys in Kafka can be used for interesting things such as Log Compaction. See Partitioning for more information.

# Set a message key; the key will be used for partitioning since no explicit
# `partition_key` is set.
kafka.deliver_message("Hello, World!", key: "hello", topic: "greetings")

Efficiently Producing Messages

While #deliver_message works fine for infrequent writes, there are a number of downsides:

• Kafka is optimized for transmitting messages in batches rather than individually, so there's a significant overhead and performance penalty in using the single-message API.
• The message delivery can fail in a number of different ways, but this simplistic API does not provide automatic retries.
• The message is not buffered, so if there is an error, it is lost.

The Producer API solves all these problems and more:

# Instantiate a new producer.
producer = kafka.producer

# Add a message to the producer buffer.
producer.produce("hello1", topic: "test-messages")

# Deliver the messages to Kafka.
producer.deliver_messages

#produce will buffer the message in the producer but will not actually send it to the Kafka cluster. Buffered messages are only delivered to the Kafka cluster once #deliver_messages is called. Since messages may be destined for different partitions, this could involve writing to more than one Kafka broker. Note that a failure to send all buffered messages after the configured number of retries will result in Kafka::DeliveryFailed being raised. This can be rescued and ignored; the messages will be kept in the buffer until the next attempt.

Read the docs for Kafka::Producer for more details.

Asynchronously Producing Messages

A normal producer will block while #deliver_messages is sending messages to Kafka, possibly for tens of seconds or even minutes at a time, depending on your timeout and retry settings. Furthermore, you have to call #deliver_messages manually, with a frequency that balances batch size with message delay.

In order to avoid blocking during message deliveries you can use the asynchronous producer API. It is mostly similar to the synchronous API, with calls to #produce and #deliver_messages. The main difference is that rather than blocking, these calls will return immediately. The actual work will be done in a background thread, with the messages and operations being sent from the caller over a thread safe queue.

# `#async_producer` will create a new asynchronous producer.
producer = kafka.async_producer

# The `#produce` API works as normal.
producer.produce("hello", topic: "greetings")

# `#deliver_messages` will return immediately.
producer.deliver_messages

# Make sure to call `#shutdown` on the producer in order to avoid leaking
# resources. `#shutdown` will wait for any pending messages to be delivered
# before returning.
producer.shutdown

By default, the delivery policy will be the same as for a synchronous producer: only wh