Add messaging.message.body.size in spans of producers writing to kafka

[shivanshuraj1333]

Is your feature request related to a problem? Please describe.

Java instrumenter (for kafka client instrumentation) uses the same attributesExtractor for both producer and consumer spans, yet messaging.message.body.size is not present for producers, the attribute list is here. (I've highlighted the diff)

The bytes attribute is extracted from for producers

opentelemetry-java-instrumentation/instrumentation/kafka/kafka-clients/kafka-clients-common/library/src/main/java/io/opentelemetry/instrumentation/kafka/internal/KafkaConsumerAttributesGetter.java

Line 52 in 8b686b5

public Long getMessageBodySize(KafkaProcessRequest request) {

But for producers, kafka client library is not instrumented and returns null from the request.

opentelemetry-java-instrumentation/instrumentation/kafka/kafka-clients/kafka-clients-common/library/src/main/java/io/opentelemetry/instrumentation/kafka/internal/KafkaProducerAttributesGetter.java

Line 36 in 8b686b5

public String getDestinationTemplate(KafkaProducerRequest request) {

---producer attributes related with messaging queues---
  {Key: "messaging.kafka.message.offset", Value: 143375.0},
  {Key: "messaging.kafka.message.key", Value: "key1"},
  {Key: "messaging.destination.partition.id", Value: "2"},
  {Key: "messaging.client_id", Value: "producer-1"},
  {Key: "messaging.system", Value: "kafka"},
  {Key: "messaging.operation", Value: "publish"},
  {Key: "messaging.destination.name", Value: "topic2"},

---producer attributes related with messaging queues---
-> {Key: "messaging.message.body.size", Value: 14.0},
  {Key: "messaging.kafka.message.offset", Value: 85405.0},
  {Key: "messaging.destination.partition.id", Value: "2"},
  {Key: "messaging.operation", Value: "process"},
  {Key: "messaging.client_id", Value: "consumer-cg1-1"},
  {Key: "messaging.system", Value: "kafka"},
-> {Key: "messaging.kafka.consumer.group", Value: "cg1"},
  {Key: "messaging.destination.name", Value: "topic1"},
  {Key: "messaging.kafka.message.key", Value: "key1"},
---

Describe the solution you'd like

Add frame work support for kafka client >= 2.6 to get the bytes in producer spans as well.framework

Describe alternatives you've considered

The data is present in the MBean object but it should also be present in producer spans as well according to messaging semconv for spans

Additional context

Details in the Slack thread here

[trask]

just a note, this attribute recently became opt-in: open-telemetry/semantic-conventions#1442