Merge pull request #70 from rtyler/proto-test

Add a repro case and fix for binary data through Hermann

Gemfile

@@ -12,4 +12,7 @@ group :test do
   gem 'rspec', '~> 3.0.0'
   gem 'rspec-its'
   gem 'system_timer', :platform => :mri_18
+  # Used for testing encoding protobufs in an out of Hermann in integration
+  # tests
+  gem 'protobuffy'
 end

Rakefile

@@ -11,21 +11,25 @@ Rake::ExtensionTask.new do |t|
   t.gem_spec = Gem::Specification.load('hermann.gemspec')
 end
 
-RSpec::Core::RakeTask.new(:spec) do |r|
-  options = ['--tag ~type:integration']
-
+def add_rspec_options(options)
   if RUBY_PLATFORM == 'java'
     options << '--tag ~platform:mri'
   else
     options << '--tag ~platform:java'
   end
+  return options
+end
+
+RSpec::Core::RakeTask.new(:spec) do |r|
+  options = add_rspec_options(['--tag ~type:integration'])
 
   r.rspec_opts = options.join(' ')
 end
 
 namespace :spec do
   RSpec::Core::RakeTask.new(:integration) do |r|
-    r.rspec_opts = '--tag type:integration'
+    options = add_rspec_options(['--tag type:integration'])
+    r.rspec_opts = options.join(' ')
   end
 end
 

lib/hermann/discovery/zookeeper.rb

@@ -24,8 +24,7 @@ module Hermann
       #    tickTime (as set in the server configuration) and a maximum
       #    of 20 times the tickTime2 times the tick time set on server"
       #
-      # @return [String] comma separated list of brokers
-      #
+      # @return [Array] List of brokers from ZK
       # @raises [NoBrokersError] if could not discover brokers thru zookeeper
       def get_brokers(timeout=0)
         brokers = []
@@ -35,7 +34,7 @@ module Hermann
         if brokers.empty?
           raise Hermann::Errors::NoBrokersError
         end
-        brokers.join(',')
+        return brokers
       end
 
       private
@@ -82,4 +81,4 @@ module Hermann
         end
     end
   end
-end
+end

lib/hermann/errors.rb

@@ -1,8 +1,7 @@
 
 module Hermann
   module Errors
-    # Error for connectivity problems with the Kafka brokers
-    class ConnectivityError < StandardError
+    class GeneralError < StandardError
       attr_reader :java_exception
 
       # Initialize a connectivity error
@@ -11,17 +10,20 @@ module Hermann
       # @param [Hash[ options
       # @option options [Java::Lang::RuntimeException] :java_exception An
       #   underlying Java exception
-      def initialize(message, options={})
+      def initialize(message='', options={})
         super(message)
         @java_exception = options[:java_exception]
       end
     end
 
+    # Error for connectivity problems with the Kafka brokers
+    class ConnectivityError < GeneralError; end
+
     # For passing incorrect config and options to kafka
-    class ConfigurationError < StandardError; end
+    class ConfigurationError < GeneralError; end
 
     # cannot discover brokers from zookeeper
-    class NoBrokersError < StandardError; end
+    class NoBrokersError < GeneralError; end
   end
 end
 

lib/hermann/provider/java_producer.rb

@@ -13,7 +13,6 @@ module Hermann
 
       #default kafka Producer options
       DEFAULTS = {
-                    'serializer.class'      => 'kafka.serializer.StringEncoder',
                     'partitioner.class'     => 'kafka.producer.DefaultPartitioner',
                     'request.required.acks' => '1',
                     'message.send.max.retries' => '0'
@@ -46,12 +45,15 @@ module Hermann
       #   will be set
       def push_single(msg, topic, unused)
         Concurrent::Promise.execute {
-          data = ProducerUtil::KeyedMessage.new(topic, msg)
+          data = ProducerUtil::KeyedMessage.new(topic, msg.to_java_bytes)
           begin
             @producer.send(data)
           rescue Java::KafkaCommon::FailedToSendMessageException => jexc
             raise Hermann::Errors::ConnectivityError.new(jexc.message,
                                                          :java_exception => jexc)
+          rescue => e
+            raise Hermann::Errors::GeneralError.new(e.message,
+                                                         :java_exception => e)
           end
         }
       end

lib/hermann/provider/java_simple_consumer.rb

@@ -60,7 +60,8 @@ module Hermann
           stream = get_stream(topic)
           it = stream.iterator
           while it.hasNext do
-            yield it.next.message.to_s
+            message = it.next.message
+            yield String.from_java_bytes(message)
           end
         rescue Exception => e
           puts "#{self.class.name}#consume exception: #{e.class.name}"

spec/discovery/zookeeper_spec.rb

@@ -18,7 +18,7 @@ describe Hermann::Discovery::Zookeeper do
     context 'with valid brokers' do
       let(:brokers) { broker_array }
       it 'gets valid string' do
-        expect(subject.get_brokers).to eq 'f:1,a:2'
+        expect(subject.get_brokers).to eq broker_array
       end
     end
     context 'with no brokers' do

spec/fixtures/integration.yml.example

@@ -1,4 +1,5 @@
 kafka:
-  brokers: "localhost:12345"
+  brokers:
+    - "localhost:12345"
   topic: "testing-topic"
-zookeepers: "localhost:2181"
+zookeepers: "localhost:2181"

spec/fixtures/testevent.pb.rb

@@ -0,0 +1,35 @@
+##
+# This file is auto-generated. DO NOT EDIT!
+#
+require 'protobuf/message'
+
+module Hermann
+
+  ##
+  # Enum Classes
+  #
+  class States < ::Protobuf::Enum
+    define :FULFILLED, 1
+    define :UNFULFILLED, 2
+    define :PENDING, 3
+    define :REJECTED, 4
+  end
+
+
+  ##
+  # Message Classes
+  #
+  class TestEvent < ::Protobuf::Message; end
+
+
+  ##
+  # Message Fields
+  #
+  class TestEvent
+    required :string, :name, 1
+    required ::Hermann::States, :state, 2
+    optional :int32, :bogomips, 3
+  end
+
+end
+

spec/fixtures/testevent.proto

@@ -0,0 +1,16 @@
+package hermann;
+
+// Generate Ruby stubs with:  protoc --ruby_out=. spec/fixtures/testevent.proto 
+
+enum States {
+  FULFILLED = 1;
+  UNFULFILLED = 2;
+  PENDING = 3;
+  REJECTED = 4;
+}
+
+message TestEvent {
+  required string name = 1;
+  required States state = 2;
+  optional int32 bogomips = 3;
+}

spec/integration/producer_spec.rb

@@ -5,39 +5,67 @@ require 'hermann/consumer'
 require 'hermann/discovery/zookeeper'
 require 'concurrent'
 
+require 'protobuf'
+require_relative '../fixtures/testevent.pb'
 
-if Hermann.jruby?
+describe 'producer' do
+  include_context 'integration test context'
 
-  class ConsumerTest
-    def create_consumer
-      zookeeper = "localhost:2181"
-      groupId   = "group1"
-      topic     = 'test'
-
-      consumer = Hermann::Consumer.new(topic, groupId, zookeeper)
-
-      consumer.consume(topic) do |msg|
-        if msg == 'msg'
-          consumer.shutdown
-          return true
-        end
+  let(:timeout) { 10 }
+  let(:message)   { 'msg' }
+  let(:consumer) do
+    Hermann::Consumer.new(topic, "rspec-group", zookeepers)
+  end
+  let(:consumer_promise) do
+    Concurrent::Promise.execute do
+      value = nil
+      puts "consuming off `#{topic}`"
+      consumer.consume(topic) do |dequeued|
+        puts "received the message: #{dequeued.inspect}"
+        value = dequeued
+        consumer.shutdown
       end
-      false
+      # Return this out of the block
+      next value
+    end
+  end
+  let(:brokers) do
+    broker_ids = Hermann::Discovery::Zookeeper.new(zookeepers).get_brokers
+    puts "using ZK discovered brokers: #{broker_ids}"
+    broker_ids
+  end
+  let(:producer) { Hermann::Producer.new(nil, brokers) }
+
+
+
+  it 'produces and consumes messages', :type => :integration, :platform => :java do
+    producer.push(message, :topic => topic).value!(timeout)
+    expect(consumer_promise.value!(timeout)).to eql(message)
+  end
+
+
+  context 'with binary data', :type => :integration, :platform => :java do
+    let(:event) do
+      Hermann::TestEvent.new(:name => 'rspec',
+                             :state => 3,
+                            :bogomips => 9001)
+    end
+
+    let(:message) { event.encode }
+
+    it 'should be a thing' do
+      producer.push(message, :topic => topic).value!(timeout)
+      dequeued = consumer_promise.value!(timeout)
+      expect(dequeued).to eql(message)
+
+      expect {
+        Hermann::TestEvent.decode(dequeued)
+      }.not_to raise_error
     end
   end
 
-  describe 'producer' do
-    include_context 'integration test context'
-
-    let(:message)   { 'msg' }
-
-    it 'produces and consumes messages', :type => :integration, :platform => :java do
-      test_consumer = Concurrent::Promise.execute { ConsumerTest.new.create_consumer }
-      broker_ids = Hermann::Discovery::Zookeeper.new(zookeepers).get_brokers
-      producer = Hermann::Producer.new(nil, broker_ids)
-      producer.push(message, :topic => topic).wait(1)
-      expect(test_consumer.value(1)).to be true
-    end
+  after :each do
+    # Make sure we shut down our connection in any case
+    consumer.shutdown
   end
-
-end
+end

spec/providers/java_producer_spec.rb

@@ -19,7 +19,7 @@ describe Hermann::Provider::JavaProducer, :platform => :java  do
     end
 
     it 'can change topic' do
-      expect(Hermann::ProducerUtil::KeyedMessage).to receive(:new).with(passed_topic, 'bar')
+      expect(Hermann::ProducerUtil::KeyedMessage).to receive(:new).with(passed_topic, anything)
       producer.push_single('bar', passed_topic, nil).wait(1)
     end
 

spec/providers/java_simple_consumer_spec.rb

@@ -17,17 +17,17 @@ describe Hermann::Provider::JavaSimpleConsumer, :platform => :java  do
   describe '#consume' do
     let(:stream)   { double }
     let(:iterator) { double }
-    let(:msg)      { double }
+    let(:msg)      { "rspec-message".to_java_bytes }
 
     it 'yields messages one at a time' do
       allow(consumer).to receive(:get_stream) { stream }
       allow(stream).to receive(:iterator) { iterator }
       allow(iterator).to receive(:hasNext).and_return(true, false)
-      allow(iterator).to receive_message_chain(:next, :message, :to_s) { msg }
+      allow(iterator).to receive_message_chain(:next, :message) { msg }
 
-      expect{ |b|
+      expect { |b|
         subject.consume(&b)
-      }.to yield_with_args(msg)
+      }.to yield_with_args(String.from_java_bytes(msg))
     end
     it 'retries consuming if there is an exception' do
       allow(consumer).to receive(:get_stream).and_raise(StandardError)