Merge pull request #2 from madebydna/aws_msk_iam_improvements

Fixing CI failures with Ruby 2.7

Gemfile

@@ -2,6 +2,3 @@ source 'https://rubygems.org'
 
 # Specify your gem's dependencies in fluent-plugin-kafka.gemspec
 gemspec
-
-gem 'json', '2.7.3' # override of 2.7.4 version
-gem 'rdkafka', ENV['RDKAFKA_VERSION_MIN_RANGE'], ENV['RDKAFKA_VERSION_MAX_RANGE'] if ENV['USE_RDKAFKA']

fluent-plugin-kafka.gemspec

@@ -19,8 +19,14 @@ Gem::Specification.new do |gem|
   gem.add_dependency "fluentd", [">= 0.10.58", "< 2"]
   gem.add_dependency 'ltsv'
   gem.add_dependency 'ruby-kafka', '>= 1.5.0', '< 2'
-  gem.add_dependency 'rdkafka'
-  gem.add_dependency 'aws-msk-iam-sasl-signer'
+
+  if ENV['USE_RDKAFKA']
+    gem.add_dependency 'rdkafka', [ENV['RDKAFKA_VERSION_MIN_RANGE'], ENV['RDKAFKA_VERSION_MAX_RANGE']]
+    if Gem::Version.new('3.0' >= Gem::Version.new(RUBY_VERSION)
+      gem.add_dependency 'aws-msk-iam-sasl-signer', '~> 0.1.1'
+    end
+  end
+
   gem.add_development_dependency "rake", ">= 0.9.2"
   gem.add_development_dependency "test-unit", ">= 3.0.8"
   gem.add_development_dependency "test-unit-rr", "~> 1.0"

lib/fluent/plugin/out_rdkafka2.rb

@@ -4,7 +4,6 @@
 require 'fluent/plugin/kafka_plugin_util'
 
 require 'rdkafka'
-require 'aws_msk_iam_sasl_signer'
 
 begin
   rdkafka_version = Gem::Version::create(Rdkafka::VERSION)
@@ -21,6 +20,10 @@
   raise "unable to patch rdkafka."
 end
 
+if Gem::Version.create(RUBY_VERSION) >= Gem::Version.create('3.0')
+  require 'aws-msk-iam-sasl-signer'
+end
+
 module Fluent::Plugin
   class Fluent::Rdkafka2Output < Output
     Fluent::Plugin.register_output('rdkafka2', self)
@@ -208,7 +211,6 @@ def add(level, message = nil)
           end
         end
       }
-      # HERE -----------------
       Rdkafka::Config.logger = log
       config = build_config
       @rdkafka = Rdkafka::Config.new(config)
@@ -217,7 +219,6 @@ def add(level, message = nil)
       if config[:"security.protocol"] == "sasl_ssl" && config[:"sasl.mechanisms"] == "OAUTHBEARER"
         Rdkafka::Config.oauthbearer_token_refresh_callback = method(:refresh_token)
       end
-      # HERE -----------------
 
       if @default_topic.nil?
         if @use_default_for_unknown_topic || @use_default_for_unknown_partition_error