improve object allocations (#456)

Author: mensfeld

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## 0.16.0 (Unreleased)
 - **[Breaking]** Retire support for Ruby 2.7.
+- **[Breaking]** Messages without headers returned by `#poll` contain frozen empty hash.
+- **[Breaking]** `HashWithSymbolKeysTreatedLikeStrings` has been removed so headers are regular hashes with string keys.
 - **[Feature]** Support incremental config describe + alter API.
 - **[Feature]** Oauthbearer token refresh callback (bruce-szalwinski-he)
 - **[Feature]** Provide ability to use topic config on a producer for custom behaviors per dispatch.
@@ -10,6 +12,7 @@
 - [Enhancement] Replace time poll based wait engine with an event based to improve response times on blocking operations and wait (nijikon + mensfeld)
 - [Enhancement] Allow for usage of the second regex engine of librdkafka by setting `RDKAFKA_DISABLE_REGEX_EXT` during build (mensfeld)
 - [Enhancement] name polling Thread as `rdkafka.native_kafka#<name>` (nijikon)
+- [Enhancement] Save two objects on message produced and lower CPU usage on message produced with small improvements.
 - [Change] Allow for native kafka thread operations deferring and manual start for consumer, producer and admin.
 - [Change] The `wait_timeout` argument in `AbstractHandle.wait` method is deprecated and will be removed in future versions without replacement. We don't rely on it's value anymore (nijikon)
 - [Fix] Background logger stops working after forking causing memory leaks (mensfeld)

lib/rdkafka/consumer/headers.rb

@@ -4,17 +4,7 @@ module Rdkafka
   class Consumer
     # Interface to return headers for a consumer message
     module Headers
-      class HashWithSymbolKeysTreatedLikeStrings < Hash
-        def [](key)
-          if key.is_a?(Symbol)
-            Kernel.warn("rdkafka deprecation warning: header access with Symbol key #{key.inspect} treated as a String. " \
-                        "Please change your code to use String keys to avoid this warning. Symbol keys will break in version 1.")
-            super(key.to_s)
-          else
-            super
-          end
-        end
-      end
+      EMPTY_HEADERS = {}.freeze
 
       # Reads a librdkafka native message's headers and returns them as a Ruby Hash
       #
@@ -28,7 +18,7 @@ def self.from_native(native_message)
         err = Rdkafka::Bindings.rd_kafka_message_headers(native_message, headers_ptrptr)
 
         if err == Rdkafka::Bindings::RD_KAFKA_RESP_ERR__NOENT
-          return {}
+          return EMPTY_HEADERS
         elsif err != Rdkafka::Bindings::RD_KAFKA_RESP_ERR_NO_ERROR
           raise Rdkafka::RdkafkaError.new(err, "Error reading message headers")
         end
@@ -39,7 +29,7 @@ def self.from_native(native_message)
         value_ptrptr = FFI::MemoryPointer.new(:pointer)
         size_ptr = Rdkafka::Bindings::SizePtr.new
 
-        headers = HashWithSymbolKeysTreatedLikeStrings.new
+        headers = {}
 
         idx = 0
         loop do

lib/rdkafka/producer.rb

@@ -319,6 +319,7 @@ def produce(
 
       delivery_handle = DeliveryHandle.new
       delivery_handle.label = label
+      delivery_handle.topic = topic
       delivery_handle[:pending] = true
       delivery_handle[:response] = -1
       delivery_handle[:partition] = -1
@@ -342,7 +343,7 @@ def produce(
           args << :int << Rdkafka::Bindings::RD_KAFKA_VTYPE_HEADER
           args << :string << key
           args << :pointer << value
-          args << :size_t << value.bytes.size
+          args << :size_t << value.bytesize
         end
       end
 

lib/rdkafka/producer/delivery_handle.rb

@@ -14,6 +14,10 @@ class DeliveryHandle < Rdkafka::AbstractHandle
       # @return [Object, nil] label set during message production or nil by default
       attr_accessor :label
 
+      # @return [String] topic where we are trying to send the message
+      # We use this instead of reading from `topic_name` pointer to save on memory allocations
+      attr_accessor :topic
+
       # @return [String] the name of the operation (e.g. "delivery")
       def operation_name
         "delivery"
@@ -26,7 +30,7 @@ def create_result
           self[:offset],
           # For part of errors, we will not get a topic name reference and in cases like this
           # we should not return it
-          self[:topic_name].null? ? nil : self[:topic_name].read_string,
+          topic,
           self[:response] != 0 ? RdkafkaError.new(self[:response]) : nil,
           label
         )

spec/rdkafka/consumer/headers_spec.rb

@@ -50,11 +50,8 @@
       expect(subject['version']).to eq("2.1.3")
     end
 
-    it 'allows Symbol key, but warns' do
-      expect(Kernel).to \
-        receive(:warn).with("rdkafka deprecation warning: header access with Symbol key :version treated as a String. " \
-                            "Please change your code to use String keys to avoid this warning. Symbol keys will break in version 1.")
-      expect(subject[:version]).to eq("2.1.3")
+    it 'does not support symbols mappings' do
+      expect(subject.key?(:version)).to eq(false)
     end
   end
 end

spec/rdkafka/producer/delivery_handle_spec.rb

@@ -9,7 +9,7 @@
       handle[:response] = response
       handle[:partition] = 2
       handle[:offset] = 100
-      handle[:topic_name] = FFI::MemoryPointer.from_string("produce_test_topic")
+      handle.topic = "produce_test_topic"
     end
   end