Use a finalizer bound to the object that will be finalized

Previous object finalizer was bound to a string (`@id`) being created on
initialization instead of the object itself.
This aimed to remove a warning about object finalizer reference after
reading [1] but changed the first argument instead of the second that is
mentioned on [1]: `proc { close }` has binding to the own object.

The current implementation removes the warning but still does not work
to run the finalizer when the reference to the producer is removed (like
`producer = nil`) therefore running only on exit. This behavior is
explained on [2].

Ref:
  1. https://www.mikeperham.com/2010/02/24/the-trouble-with-ruby-finalizers/
  2. https://ruby-doc.org/core-3.0.2/ObjectSpace.html#method-c-define_finalizer

---

Changelog: fix
Ref: https://www.mikeperham.com/2010/02/24/the-trouble-with-ruby-finalizers/
Ref: https://ruby-doc.org/core-3.0.2/ObjectSpace.html#method-c-define_finalizer
See-also: https://www.mikeperham.com/2010/02/24/the-trouble-with-ruby-finalizers/
See-also: https://ruby-doc.org/core-3.0.2/ObjectSpace.html#method-c-define_finalizer

Author: jvortmann

lib/rdkafka/producer.rb

@@ -1,4 +1,4 @@
-require "securerandom"
+require "objspace"
 
 module Rdkafka
   # A producer for Kafka messages. To create a producer set up a {Config} and call {Config#producer producer} on that.
@@ -9,14 +9,31 @@ class Producer
     # @return [Proc, nil]
     attr_reader :delivery_callback
 
+    class ClientFinalizer
+      def initialize(client, polling_thread, reference_destroyer)
+        @client = client
+        @polling_thread = polling_thread
+        @reference_destroyer = reference_destroyer
+      end
+
+      def call(object_id)
+        return unless @client
+
+        # Indicate to polling thread that we're closing
+        @polling_thread[:closing] = true
+        # Wait for the polling thread to finish up
+        @polling_thread.join
+
+        Rdkafka::Bindings.rd_kafka_destroy(@client)
+
+        @reference_destroyer.call
+      end
+    end
+
     # @private
     def initialize(native_kafka)
-      @id = SecureRandom.uuid
       @native_kafka = native_kafka
 
-      # Makes sure, that the producer gets closed before it gets GCed by Ruby
-      ObjectSpace.define_finalizer(@id, proc { close })
-
       # Start thread to poll client for delivery callbacks
       @polling_thread = Thread.new do
         loop do
@@ -29,6 +46,9 @@ def initialize(native_kafka)
       end
       @polling_thread.abort_on_exception = true
       @polling_thread[:closing] = false
+
+      # Makes sure, that the producer gets closed before it gets GCed by Ruby
+      ObjectSpace.define_finalizer(self, ClientFinalizer.new(@native_kafka, @polling_thread, proc { @native_kafka = nil }))
     end
 
     # Set a callback that will be called every time a message is successfully produced.
@@ -44,16 +64,9 @@ def delivery_callback=(callback)
 
     # Close this producer and wait for the internal poll queue to empty.
     def close
-      ObjectSpace.undefine_finalizer(@id)
-
-      return unless @native_kafka
+      ObjectSpace.undefine_finalizer(self)
 
-      # Indicate to polling thread that we're closing
-      @polling_thread[:closing] = true
-      # Wait for the polling thread to finish up
-      @polling_thread.join
-      Rdkafka::Bindings.rd_kafka_destroy(@native_kafka)
-      @native_kafka = nil
+      ClientFinalizer.new(@native_kafka, @polling_thread, proc { @native_kafka = nil }).call(self.object_id)
     end
 
     # Partition count for a given topic.