topic_from_attribute does not work properly

[yilmazo]

Component(s)

exporter/kafka

What happened?

Description

When using topic_from_attribute feature on kafka exporter alongside with the batcher processor, even when there are logs to send to different topics in a single batch, exporter selects topic from one message and sends all messages to that topic.
Relevant code can be seen here.
I tested this only with Logs, but based on the code on the link, I assume this issue occurs with all telemetry types.

Steps to Reproduce

• Prepare a collector pipeline with batcher processor and kafka exporter
• Use topic_from_attribute=kafka.topic.name on kafka exporter
• Send Logs to collector with different topic values on their relevant attribute

Expected Result

Each message is sent to topics according to their kafka.topic.name resource attribute.

Actual Result

Logs in a single batch (a single kafka message) have different resource attributes (and different kafka.topic.name values). As a result, some messages are sent to wrong topics.

Collector version

v0.116.0

Environment information

No response

OpenTelemetry Collector configuration

receivers:
  otlp:
    protocols:
      grpc: 
      http:

processors:
  batch/logging:
    send_batch_size: 64
    send_batch_max_size: 128
    timeout: 30s

exporters:
  kafka/logging:
    brokers:
    - kafka-poc-kafka-brokers.kafka:9092
    encoding: otlp_proto
    producer:
      compression: zstd
      max_message_bytes: 10000000
      required_acks: 1
    protocol_version: 3.5.0
    retry_on_failure:
      enabled: true
      initial_interval: 10s
      max_elapsed_time: 20s
      max_interval: 10s
    sending_queue:
      enabled: true
      num_consumers: 32
      queue_size: 1000
    timeout: 10s
    topic_from_attribute: kafka.topic.name
    topic: otelcol-log.dc1.default.1

service:
  pipelines:
    logs/logging:
      receivers:
      - otlp
      processors:
      - batch/logging
      exporters:
      - kafka/logging

Log output

Additional context

No response

[github-actions]

Pinging code owners:

• exporter/kafka: @pavolloffay @MovieStoreGuy

See Adding Labels via Comments if you do not have permissions to add labels yourself.