Revert "fix: delete each_batch method from targets of instrumentation"

This reverts commit de95123.

Author: joker1007

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

@@ -42,6 +42,34 @@ def each
             end
           end
 
+          # each_batch method is deleted in rdkafka 0.20.0
+          if Gem::Version.new(::Rdkafka::VERSION) < Gem::Version.new('0.20.0')
+            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
+          end
+
           private
 
           def tracer

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

@@ -165,6 +165,74 @@
         begin; consumer&.close; rescue StandardError; end
       end
     end
+
+    # each_batch method is deleted in rdkafka 0.20.0
+    if Gem::Version.new(Rdkafka::VERSION) < Gem::Version.new('0.20.0')
+      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
 end