From bc3d0f055079b86de189007b82724973f9e240b7 Mon Sep 17 00:00:00 2001 From: Alexander Mossin Date: Tue, 29 Nov 2011 20:38:54 +0100 Subject: [PATCH 1/7] Moved encoding declaration to the top, where it is actually used by ruby --- spec/producer_spec.rb | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/spec/producer_spec.rb b/spec/producer_spec.rb index 1cfb933..e6586b6 100644 --- a/spec/producer_spec.rb +++ b/spec/producer_spec.rb @@ -1,3 +1,5 @@ +# encoding: utf-8 + # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. @@ -12,7 +14,6 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -# encoding: utf-8 require File.dirname(__FILE__) + '/spec_helper' describe Producer do From 9d6825f6ae62f98a810b862a0dce17ae572bd943 Mon Sep 17 00:00:00 2001 From: Alexander Mossin Date: Tue, 29 Nov 2011 20:41:34 +0100 Subject: [PATCH 2/7] Fix failing spec - force UTF8 encoding on ruby 1.9+. Specs now pass with 1.8.7, 1.9.2 and JRuby --- spec/producer_spec.rb | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/spec/producer_spec.rb b/spec/producer_spec.rb index e6586b6..fad615e 100644 --- a/spec/producer_spec.rb +++ b/spec/producer_spec.rb @@ -65,11 +65,13 @@ message = Kafka::Message.new("ümlaut") encoded = @producer.encode(message) data = [encoded.size].pack("N") + encoded + payload = Kafka::Message.parse_from(data).payload + if RUBY_VERSION[0,3] == "1.8" # Use old iconv on Ruby 1.8 for encoding ic = Iconv.new('UTF-8//IGNORE', 'UTF-8') - ic.iconv(Kafka::Message.parse_from(data).payload).should eql("ümlaut") + ic.iconv(payload).should eql("ümlaut") else - Kafka::Message.parse_from(data).payload.force_encoding(Encoding::ASCII_8BIT).should eql("ümlaut") + payload.force_encoding(Encoding::UTF_8).should eql("ümlaut") end end end From b4d9a4834f8520b1b6f5d9cd280d6857bd6820d7 Mon Sep 17 00:00:00 2001 From: Alexander Mossin Date: Tue, 29 Nov 2011 20:50:33 +0100 Subject: [PATCH 3/7] Added .gitignore --- .gitignore | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 .gitignore diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7b0c1fc --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +.rvmrc + From f8058cdd70f3ba6b41b0c100da0bd145c7ecb10b Mon Sep 17 00:00:00 2001 From: Alexander Mossin Date: Tue, 29 Nov 2011 20:50:48 +0100 Subject: [PATCH 4/7] Version bump --- kafka-rb.gemspec | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/kafka-rb.gemspec b/kafka-rb.gemspec index b4bf039..357f016 100644 --- a/kafka-rb.gemspec +++ b/kafka-rb.gemspec @@ -16,7 +16,7 @@ Gem::Specification.new do |s| s.name = %q{kafka-rb} - s.version = "0.0.6" + s.version = "0.0.7" s.required_rubygems_version = Gem::Requirement.new(">= 0") if s.respond_to? :required_rubygems_version= s.authors = ["Alejandro Crosa"] From a49d861d567ba730bfa51d5bd233097b09cea935 Mon Sep 17 00:00:00 2001 From: Alexander Mossin Date: Tue, 29 Nov 2011 21:07:22 +0100 Subject: [PATCH 5/7] Added zookeeper gem dependency --- kafka-rb.gemspec | 2 ++ 1 file changed, 2 insertions(+) diff --git a/kafka-rb.gemspec b/kafka-rb.gemspec index 357f016..0331060 100644 --- a/kafka-rb.gemspec +++ b/kafka-rb.gemspec @@ -43,4 +43,6 @@ Gem::Specification.new do |s| else s.add_dependency(%q, [">= 0"]) end + + s.add_dependency(%q, [">= 0.4.4"]) end From b0c1cb47d2da69e89908e3f2bbd226f3c079a281 Mon Sep 17 00:00:00 2001 From: Alexander Mossin Date: Tue, 29 Nov 2011 23:01:48 +0100 Subject: [PATCH 6/7] Added consumer and group id --- lib/kafka/consumer.rb | 12 +++++++++++- spec/consumer_spec.rb | 11 +++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/lib/kafka/consumer.rb b/lib/kafka/consumer.rb index c273ef5..c2e69a0 100644 --- a/lib/kafka/consumer.rb +++ b/lib/kafka/consumer.rb @@ -20,8 +20,9 @@ class Consumer CONSUME_REQUEST_TYPE = Kafka::RequestType::FETCH MAX_SIZE = 1048576 # 1 MB DEFAULT_POLLING_INTERVAL = 2 # 2 seconds + DEFAULT_GROUP_ID = "test" - attr_accessor :topic, :partition, :offset, :max_size, :request_type, :polling + attr_accessor :topic, :partition, :offset, :max_size, :request_type, :polling, :group_id, :consumer_id def initialize(options = {}) self.topic = options[:topic] || "test" @@ -32,6 +33,9 @@ def initialize(options = {}) self.max_size = options[:max_size] || MAX_SIZE self.request_type = options[:request_type] || CONSUME_REQUEST_TYPE self.polling = options[:polling] || DEFAULT_POLLING_INTERVAL + self.group_id = options[:group_id] || DEFAULT_GROUP_ID + self.consumer_id = options[:consumer_id] || Consumer.generate_id(self.group_id) + self.connect(self.host, self.port) end @@ -94,5 +98,11 @@ def parse_message_set_from(data) self.offset += processed messages end + + private + + def self.generate_id(group_id) + "#{group_id}_#{Socket.gethostname}-#{Time.now.usec}" + end end end diff --git a/spec/consumer_spec.rb b/spec/consumer_spec.rb index 8f411e5..be8e2f4 100644 --- a/spec/consumer_spec.rb +++ b/spec/consumer_spec.rb @@ -64,6 +64,17 @@ Consumer::MAX_SIZE.should eql(1048576) @consumer.max_size.should eql(1048576) end + + it "should generate a consumer id on initialize" do + Time.stub!(:now).and_return(Time.new(0)) + Socket.stub!(:gethostname).and_return("foohost") + consumer = Consumer.new({:group_id => "foogroup"}) + consumer.consumer_id.should eql("foogroup_foohost-0") + end + + it "should have a group id" do + @consumer.should respond_to(:group_id) + end it "should return the size of the request" do @consumer.request_size.should eql(24) From 769c8ab295687e5d234222e6a1bf75adae132c7c Mon Sep 17 00:00:00 2001 From: Alexander Mossin Date: Wed, 30 Nov 2011 13:45:49 +0100 Subject: [PATCH 7/7] Added ConsumerRegistry and the register method --- Rakefile | 2 +- kafka-rb.gemspec | 4 +-- lib/kafka.rb | 1 + lib/kafka/consumer.rb | 8 ++++- lib/kafka/consumer_registry.rb | 61 ++++++++++++++++++++++++++++++++++ 5 files changed, 72 insertions(+), 4 deletions(-) create mode 100644 lib/kafka/consumer_registry.rb diff --git a/Rakefile b/Rakefile index 00aa14c..ff81799 100644 --- a/Rakefile +++ b/Rakefile @@ -21,7 +21,7 @@ require 'rspec/core/rake_task' GEM = 'kafka-rb' GEM_NAME = 'Kafka Client' -GEM_VERSION = '0.0.5' +GEM_VERSION = '0.0.7' AUTHORS = ['Alejandro Crosa'] EMAIL = "alejandrocrosa@gmail.com" HOMEPAGE = "http://github.com/acrosa/kafka-rb" diff --git a/kafka-rb.gemspec b/kafka-rb.gemspec index 0331060..5df174d 100644 --- a/kafka-rb.gemspec +++ b/kafka-rb.gemspec @@ -25,7 +25,7 @@ Gem::Specification.new do |s| s.description = %q{kafka-rb allows you to produce and consume messages using the Kafka distributed publish/subscribe messaging service.} s.email = %q{alejandrocrosa@gmail.com} s.extra_rdoc_files = ["LICENSE"] - s.files = ["LICENSE", "README.md", "Rakefile", "lib/kafka", "lib/kafka/batch.rb", "lib/kafka/consumer.rb", "lib/kafka/io.rb", "lib/kafka/message.rb", "lib/kafka/producer.rb", "lib/kafka/request_type.rb", "lib/kafka/error_codes.rb", "lib/kafka.rb", "spec/batch_spec.rb", "spec/consumer_spec.rb", "spec/io_spec.rb", "spec/kafka_spec.rb", "spec/message_spec.rb", "spec/producer_spec.rb", "spec/spec_helper.rb"] + s.files = ["LICENSE", "README.md", "Rakefile", "lib/kafka", "lib/kafka/batch.rb", "lib/kafka/consumer.rb", "lib/kafka/consumer_registry.rb", "lib/kafka/io.rb", "lib/kafka/message.rb", "lib/kafka/producer.rb", "lib/kafka/request_type.rb", "lib/kafka/error_codes.rb", "lib/kafka.rb", "spec/batch_spec.rb", "spec/consumer_spec.rb", "spec/io_spec.rb", "spec/kafka_spec.rb", "spec/message_spec.rb", "spec/producer_spec.rb", "spec/spec_helper.rb"] s.homepage = %q{http://github.com/acrosa/kafka-rb} s.require_paths = ["lib"] s.rubygems_version = %q{1.3.7} @@ -44,5 +44,5 @@ Gem::Specification.new do |s| s.add_dependency(%q, [">= 0"]) end - s.add_dependency(%q, [">= 0.4.4"]) + s.add_dependency(%q, [">= 0.8.5"]) end diff --git a/lib/kafka.rb b/lib/kafka.rb index ccf49c3..cee7409 100644 --- a/lib/kafka.rb +++ b/lib/kafka.rb @@ -25,6 +25,7 @@ require File.join(File.dirname(__FILE__), "kafka", "message") require File.join(File.dirname(__FILE__), "kafka", "producer") require File.join(File.dirname(__FILE__), "kafka", "consumer") +require File.join(File.dirname(__FILE__), "kafka", "consumer_registry") module Kafka end diff --git a/lib/kafka/consumer.rb b/lib/kafka/consumer.rb index c2e69a0..e9be97d 100644 --- a/lib/kafka/consumer.rb +++ b/lib/kafka/consumer.rb @@ -22,7 +22,7 @@ class Consumer DEFAULT_POLLING_INTERVAL = 2 # 2 seconds DEFAULT_GROUP_ID = "test" - attr_accessor :topic, :partition, :offset, :max_size, :request_type, :polling, :group_id, :consumer_id + attr_accessor :topic, :partition, :offset, :max_size, :request_type, :polling, :group_id, :consumer_id, :zk_connect def initialize(options = {}) self.topic = options[:topic] || "test" @@ -35,6 +35,12 @@ def initialize(options = {}) self.polling = options[:polling] || DEFAULT_POLLING_INTERVAL self.group_id = options[:group_id] || DEFAULT_GROUP_ID self.consumer_id = options[:consumer_id] || Consumer.generate_id(self.group_id) + self.zk_connect = options[:zk_connect] + + unless self.zk_connect.nil? + @consumer_registry = Kafka::ConsumerRegistry.new(self.zk_connect) + @consumer_registry.register(self.group_id, self.consumer_id, self.topic) + end self.connect(self.host, self.port) end diff --git a/lib/kafka/consumer_registry.rb b/lib/kafka/consumer_registry.rb new file mode 100644 index 0000000..37e265e --- /dev/null +++ b/lib/kafka/consumer_registry.rb @@ -0,0 +1,61 @@ +# This class handles the consumers interaction with zookeeper +# +# Directories: +# 1. Consumer id registry: +# /consumers/[group_id]/ids[consumer_id] -> topic1,...topicN +# A consumer has a unique consumer id within a consumer group. A consumer registers its id as an ephemeral znode +# and puts all topics that it subscribes to as the value of the znode. The znode is deleted when the client is gone. +# A consumer subscribes to event changes of the consumer id registry within its group. +# +# The consumer id is picked up from configuration, instead of the sequential id assigned by ZK. Generated sequential +# ids are hard to recover during temporary connection loss to ZK, since it's difficult for the client to figure out +# whether the creation of a sequential znode has succeeded or not. More details can be found at +# (http://wiki.apache.org/hadoop/ZooKeeper/ErrorHandling) +# +# 2. Broker node registry: +# /brokers/[0...N] --> { "host" : "host:port", +# "topics" : {"topic1": ["partition1" ... "partitionN"], ..., +# "topicN": ["partition1" ... "partitionN"] } } +# This is a list of all present broker brokers. A unique logical node id is configured on each broker node. A broker +# node registers itself on start-up and creates a znode with the logical node id under /brokers. The value of the znode +# is a JSON String that contains (1) the host name and the port the broker is listening to, (2) a list of topics that +# the broker serves, (3) a list of logical partitions assigned to each topic on the broker. +# A consumer subscribes to event changes of the broker node registry. +# +# 3. Partition owner registry: +# /consumers/[group_id]/owner/[topic]/[broker_id-partition_id] --> consumer_node_id +# This stores the mapping before broker partitions and consumers. Each partition is owned by a unique consumer +# within a consumer group. The mapping is reestablished after each rebalancing. +# +# 4. Consumer offset tracking: +# /consumers/[group_id]/offsets/[topic]/[broker_id-partition_id] --> offset_counter_value +# Each consumer tracks the offset of the latest message consumed for each partition. +# +#/ + +require 'zk' +require 'json' + +module Kafka + class ConsumerRegistry + def initialize(connect='localhost:9092') + @connect = connect + end + + def register(group_id, consumer_id, topic) + dir = "/consumers/#{group_id}/ids" + keeper.mkdir_p(dir) + + keeper.create( + "#{dir}/#{consumer_id}", + {topic => 1}.to_json, + {:ephemeral => true} + ) + end + + def keeper + @keeper = ZK.new(@connect) if @keeper.nil? + @keeper + end + end +end \ No newline at end of file