feat: update gem version requirement of rdkafka

instrumentation/rdkafka/lib/opentelemetry/instrumentation/rdkafka/instrumentation.rb

@@ -11,7 +11,7 @@ module Rdkafka
       class Instrumentation < OpenTelemetry::Instrumentation::Base
         compatible do
           gem_version = Gem::Version.new(::Rdkafka::VERSION)
-          Gem::Requirement.new('>= 0.10.0', '< 0.20.0').satisfied_by?(gem_version)
+          Gem::Requirement.new('>= 0.10.0', '< 0.22.0').satisfied_by?(gem_version)
         end
 
         install do |_config|

instrumentation/rdkafka/lib/opentelemetry/instrumentation/rdkafka/patches/consumer.rb

@@ -42,31 +42,6 @@ def each
             end
           end
 
-          def each_batch(max_items: 100, bytes_threshold: Float::INFINITY, timeout_ms: 250, yield_on_error: false, &block)
-            super do |messages, error|
-              if messages.empty?
-                yield messages, error
-              else
-                attributes = {
-                  'messaging.system' => 'kafka',
-                  'messaging.destination_kind' => 'topic',
-                  'messaging.kafka.message_count' => messages.size
-                }
-
-                links = messages.map do |message|
-                  trace_context = OpenTelemetry.propagation.extract(message.headers, getter: GETTER)
-                  span_context = OpenTelemetry::Trace.current_span(trace_context).context
-                  OpenTelemetry::Trace::Link.new(span_context) if span_context.valid?
-                end
-                links.compact!
-
-                tracer.in_span('batch process', attributes: attributes, links: links, kind: :consumer) do
-                  yield messages, error
-                end
-              end
-            end
-          end
-
           private
 
           def tracer

instrumentation/rdkafka/test/opentelemetry/instrumentation/rdkafka/patches/consumer_test.rb

@@ -165,71 +165,6 @@
         begin; consumer&.close; rescue StandardError; end
       end
     end
-
-    describe '#each_batch' do
-      it 'traces each_batch call' do
-        skip "#{Rdkafka::VERSION} is not supported" unless instrumentation.compatible?
-
-        rand_hash = SecureRandom.hex(10)
-        topic_name = "consumer-patch-batch-trace-#{rand_hash}"
-        config = { 'bootstrap.servers': "#{host}:#{port}" }
-
-        producer = Rdkafka::Config.new(config).producer
-        delivery_handles = []
-
-        delivery_handles << producer.produce(
-          topic: topic_name,
-          payload: 'wow',
-          key: 'Key 1'
-        )
-
-        delivery_handles << producer.produce(
-          topic: topic_name,
-          payload: 'super',
-          key: 'Key 2'
-        )
-
-        delivery_handles.each(&:wait)
-
-        consumer_config = config.merge(
-          'group.id': 'me',
-          'auto.offset.reset': 'smallest' # https://stackoverflow.com/a/51081649
-        )
-        consumer = Rdkafka::Config.new(config.merge(consumer_config)).consumer
-        consumer.subscribe(topic_name)
-
-        begin
-          consumer.each_batch(max_items: 2) do |messages|
-            raise 'oops' unless messages.empty?
-          end
-        rescue StandardError
-        end
-
-        span = spans.find { |s| s.name == 'batch process' }
-        _(span.kind).must_equal(:consumer)
-        _(span.attributes['messaging.kafka.message_count']).must_equal(2)
-
-        event = span.events.first
-        _(event.name).must_equal('exception')
-        _(event.attributes['exception.type']).must_equal('RuntimeError')
-        _(event.attributes['exception.message']).must_equal('oops')
-
-        first_link = span.links[0]
-        linked_span_context = first_link.span_context
-        _(linked_span_context.trace_id).must_equal(spans[0].trace_id)
-        _(linked_span_context.span_id).must_equal(spans[0].span_id)
-
-        second_link = span.links[1]
-        linked_span_context = second_link.span_context
-        _(linked_span_context.trace_id).must_equal(spans[1].trace_id)
-        _(linked_span_context.span_id).must_equal(spans[1].span_id)
-
-        _(spans.size).must_equal(3)
-      ensure
-        begin; producer&.close; rescue StandardError; end
-        begin; consumer&.close; rescue StandardError; end
-      end
-    end
   end
 end