No description, website, or topics provided.
Ruby Shell
Clone or download
Latest commit 74bfb89 Aug 12, 2016


Build Status

Version 2.x => Kafka

Version 3.x => Kafka

Version 4.x => Kafka


  • JRuby 1.7.19+ installed.


This gem is primarily used to wrap most of the [Kafka high level consumer] and [Kafka producer] API into jruby. The [Kafka Consumer Group Example] is pretty much ported to this library.

Note that the Scala Kafka::Producer will deprecate and Java Kafka::KafkaProducer is taking over. (The 0.8.2.x and 0.9x consumer/producer APIs are similar)


This package is now distributed via but you can build it using the following instructions.

From the root of the project run:

$ bundle install
$ rake setup jar package

You can run the following to install the resulting package:

$ gem install jruby-kafka*.gem

Add this line to your application's Gemfile:

gem 'jruby-kafka', "~> #{your_version_here}" # '~> 2.0.0' for brokers, '~> 3.0.0' for 0.9 brokers


If you want to run the tests, make sure you already have downloaded Kafka 0.8.X, followed the kafka quickstart instructions and have KAFKA_PATH set in the environment.


The following producer code sends a message to a test topic.

require 'jruby-kafka'

producer_options = {:broker_list => "localhost:9092", "serializer.class" => "kafka.serializer.StringEncoder"}

producer =
producer.send_msg("test", nil, "here's a test message")    

The following consumer example is the Ruby equivalent of the Kafka high-level consumer group example. It listens for 10 seconds to the test topic and prints out messages as they are received from Kafka in two threads. The test topic should have at least two partitions for each thread to receive messages.

require 'jruby-kafka'

consumer_options = {
  zookeeper_connect:  'localhost:2181',
  group_id:           'my_consumer_group',
  topic:              'test',
  num_streams:        2,
  auto_offset_reset:  "smallest"

consumer =

def consumer_test(stream, thread_num)
  it = stream.iterator
  puts "Thread #{thread_num}: #{}" while it.hasNext 
  puts "Shutting down Thread: #{thread_num}"

streams  = consumer.message_streams
streams.each_with_index do |stream, thread_num| { consumer_test stream, thread_num }

sleep 10


  1. Fork it
  2. Create your feature branch (git checkout -b my-new-feature)
  3. Commit your changes (git commit -am 'Add some feature')
  4. Push to the branch (git push origin my-new-feature)
  5. Create new Pull Request