[receiver/kafka] Remove messageHeaders abstraction since Sarama is deprecated

[paulojmdias]

Description

The messageHeaders interface, kafkaMessage interface, and related wrapper types were designed to abstract over Sarama and franz-go record types. Now that Sarama has been fully removed, and only the franz-go implementation remains, we should remove that abstraction.

Replaced kafkaMessage with *kgo.Record directly, replaced messageHeaders with []kgo.RecordHeader.

Link to tracking issue

Fixes #46816

Testing

Updated unit tests and re-run the benchmark tests to get the possible improvements with the results below.

goos: darwin
goarch: arm64
pkg: github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver
cpu: Apple M3 Pro
                                                  │ header_wrapper.txt │          header_franz.txt           │
                                                  │       sec/op       │   sec/op     vs base                │
ContextWithHeaders/no_headers-12                          2.212n ±  7%   2.111n ± 6%   -4.52% (p=0.022 n=10)
ContextWithHeaders/1_header-12                            356.9n ±  8%   231.7n ± 4%  -35.08% (p=0.000 n=10)
ContextWithHeaders/5_headers-12                           549.1n ±  8%   437.5n ± 5%  -20.32% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=1-12           62.62n ±  6%   43.75n ± 4%  -30.13% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=4-12           137.6n ±  4%   113.9n ± 5%  -17.22% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=8-12           243.2n ±  8%   215.2n ± 4%  -11.49% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=16-12          547.0n ± 10%   511.8n ± 5%   -6.44% (p=0.003 n=10)
GetMessageHeaderResourceAttributes/headers=32-12          1.546µ ±  5%   1.476µ ± 6%        ~ (p=0.093 n=10)
GetMessageHeaderResourceAttributes/headers=64-12          5.465µ ±  4%   5.349µ ± 4%   -2.13% (p=0.045 n=10)
GetMessageHeaderResourceAttributes/headers=128-12         14.89µ ±  5%   14.50µ ± 5%        ~ (p=0.631 n=10)
geomean                                                   379.8n         325.7n       -14.24%

                                                  │ header_wrapper.txt │            header_franz.txt            │
                                                  │        B/op        │     B/op      vs base                  │
ContextWithHeaders/no_headers-12                          0.000 ± 0%       0.000 ± 0%        ~ (p=1.000 n=10) ¹
ContextWithHeaders/1_header-12                            984.0 ± 0%       520.0 ± 0%  -47.15% (p=0.000 n=10)
ContextWithHeaders/5_headers-12                          1064.0 ± 0%       602.0 ± 0%  -43.42% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=1-12          32.000 ± 0%       8.000 ± 0%  -75.00% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=4-12           56.00 ± 0%       32.00 ± 0%  -42.86% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=8-12           88.00 ± 0%       64.00 ± 0%  -27.27% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=16-12          152.0 ± 0%       128.0 ± 0%  -15.79% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=32-12          280.0 ± 0%       256.0 ± 0%   -8.57% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=64-12          536.0 ± 0%       512.0 ± 0%   -4.48% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=128-12       1.225Ki ± 0%     1.191Ki ± 2%   -2.75% (p=0.000 n=10)
geomean                                                              ²                 -31.66%                ²
¹ all samples are equal
² summaries must be >0 to compute geomean

                                                  │ header_wrapper.txt │           header_franz.txt           │
                                                  │     allocs/op      │ allocs/op   vs base                  │
ContextWithHeaders/no_headers-12                          0.000 ± 0%     0.000 ± 0%        ~ (p=1.000 n=10) ¹
ContextWithHeaders/1_header-12                           11.000 ± 0%     6.000 ± 0%  -45.45% (p=0.000 n=10)
ContextWithHeaders/5_headers-12                           19.00 ± 0%     14.00 ± 0%  -26.32% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=1-12           2.000 ± 0%     1.000 ± 0%  -50.00% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=4-12           5.000 ± 0%     4.000 ± 0%  -20.00% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=8-12           9.000 ± 0%     8.000 ± 0%  -11.11% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=16-12          17.00 ± 0%     16.00 ± 0%   -5.88% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=32-12          33.00 ± 0%     32.00 ± 0%   -3.03% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=64-12          65.00 ± 0%     64.00 ± 0%   -1.54% (p=0.000 n=10)
GetMessageHeaderResourceAttributes/headers=128-12         129.0 ± 0%     128.0 ± 0%   -0.78% (p=0.000 n=10)
geomean                                                              ²               -18.62%                ²
¹ all samples are equal

[Copilot]

Pull request overview

This PR removes the now-unnecessary Kafka message/header abstraction layer in kafkareceiver after Sarama removal, using franz-go types (*kgo.Record and []kgo.RecordHeader) directly.

Changes:

• Replaced the kafkaMessage interface with *kgo.Record across receiver/consumer code paths.
• Replaced messageHeaders with []kgo.RecordHeader, updating header extraction/context propagation accordingly.
• Removed the wrapper/adapter implementation file and updated associated tests/benchmarks.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
receiver/kafkareceiver/kafka_receiver_bench_test.go	Benchmarks updated to use []kgo.RecordHeader directly.
receiver/kafkareceiver/kafka_receiver.go	Receiver pipeline now consumes *kgo.Record; header helpers updated to operate on []kgo.RecordHeader.
receiver/kafkareceiver/consumer_message.go	Removed now-obsolete message/header abstraction wrappers.
receiver/kafkareceiver/consumer_franz_test.go	Tests updated for new consumeMessageFunc signature (*kgo.Record).
receiver/kafkareceiver/consumer_franz.go	Consumer now passes *kgo.Record directly (no wrapper).

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[paulojmdias]

/rerun

[axw]

Thank you!