Merge pull request #468 from raytung/fix/rdkafka0-12-0

fix(out_rdkafka2): fix compatibility with rdkafka 0.12.0

Author: ashie

.github/workflows/linux.yml

@@ -15,7 +15,10 @@ jobs:
         ruby: [ '3.2', '3.1', '3.0', '2.7' ]
         os:
           - ubuntu-latest
-    name: Ruby ${{ matrix.ruby }} unit testing on ${{ matrix.os }}
+        rdkafka_versions:
+          - { min: '>= 0.6.0', max: '< 0.12.0' }
+          - { min: '>= 0.12.0', max: '>= 0.12.0' }
+    name: Ruby ${{ matrix.ruby }} unit testing on ${{ matrix.os }} with rdkafka gem version (min ${{ matrix.rdkafka_versions.min }} max ${{ matrix.rdkafka_versions.max }})
     steps:
     - uses: actions/checkout@v3
     - uses: ruby/setup-ruby@v1
@@ -33,6 +36,8 @@ jobs:
     - name: unit testing
       env:
         CI: true
+        RDKAFKA_VERSION_MIN_RANGE: ${{ matrix.rdkafka_versions.min }}
+        RDKAFKA_VERSION_MAX_RANGE: ${{ matrix.rdkafka_versions.max }}
       run: |
         sudo ./ci/prepare-kafka-server.sh
         gem install bundler rake

Gemfile

@@ -3,4 +3,4 @@ source 'https://rubygems.org'
 # Specify your gem's dependencies in fluent-plugin-kafka.gemspec
 gemspec
 
-gem 'rdkafka', '>= 0.6.0' if ENV["USE_RDKAFKA"]
+gem 'rdkafka', ENV['RDKAFKA_VERSION_MIN_RANGE'], ENV['RDKAFKA_VERSION_MAX_RANGE'] if ENV['USE_RDKAFKA']

lib/fluent/plugin/out_rdkafka2.rb

@@ -5,9 +5,37 @@
 
 require 'rdkafka'
 
+# This is required for `rdkafka` version >= 0.12.0
+# Overriding the close method in order to provide a time limit for when it should be forcibly closed
+class Rdkafka::Producer::Client
+  # return false if producer is forcefully closed, otherwise return true
+  def close(timeout=nil)
+    return unless @native
+
+    # Indicate to polling thread that we're closing
+    @polling_thread[:closing] = true
+    # Wait for the polling thread to finish up
+    thread = @polling_thread.join(timeout)
+
+    Rdkafka::Bindings.rd_kafka_destroy(@native)
+
+    @native = nil
+
+    return !thread.nil?
+  end
+end
+
 class Rdkafka::Producer
   # return false if producer is forcefully closed, otherwise return true
   def close(timeout = nil)
+    rdkafka_version = Rdkafka::VERSION || '0.0.0'
+    # Rdkafka version >= 0.12.0 changed its internals
+    if Gem::Version::create(rdkafka_version) >= Gem::Version.create('0.12.0')
+      ObjectSpace.undefine_finalizer(self)
+
+      return @client.close(timeout)
+    end
+
     @closing = true
     # Wait for the polling thread to finish up
     # If the broker isn't alive, the thread doesn't exit