Skip to content

Commit

Permalink
fix: suppress deprecation warning in Rdkafka Instrumentation
Browse files Browse the repository at this point in the history
  • Loading branch information
yoheyk committed Feb 21, 2024
1 parent c95e919 commit 369fb4a
Showing 1 changed file with 10 additions and 2 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ def each
message_key = extract_message_key(message.key)
attributes['messaging.kafka.message_key'] = message_key if message_key

parent_context = OpenTelemetry.propagation.extract(message.headers, getter: OpenTelemetry::Common::Propagation.symbol_key_getter)
parent_context = OpenTelemetry.propagation.extract(message.headers, getter: getter)
span_context = OpenTelemetry::Trace.current_span(parent_context).context
links = [OpenTelemetry::Trace::Link.new(span_context)] if span_context.valid?

Expand All @@ -47,7 +47,7 @@ def each_batch(max_items: 100, bytes_threshold: Float::INFINITY, timeout_ms: 250
}

links = messages.map do |message|
trace_context = OpenTelemetry.propagation.extract(message.headers, getter: OpenTelemetry::Common::Propagation.symbol_key_getter)
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
Expand All @@ -74,6 +74,14 @@ def extract_message_key(key)
rescue Encoding::UndefinedConversionError
nil
end

def getter
if Gem::Version.new(::Rdkafka::VERSION) >= Gem::Version.new('0.13.0')
Context::Propagation.text_map_getter
else
OpenTelemetry::Common::Propagation.symbol_key_getter
end
end
end
end
end
Expand Down

0 comments on commit 369fb4a

Please sign in to comment.