BatchKafkaTelemetrySettingsProvider

def get_publish_attrs_from_kwargs(
    self,
    kwargs: "AnyDict",
) -> "AnyDict":
    attrs = {
        SpanAttributes.MESSAGING_SYSTEM: self.messaging_system,
        SpanAttributes.MESSAGING_DESTINATION_NAME: kwargs["topic"],
        SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: kwargs["correlation_id"],
    }

    if (partition := kwargs.get("partition")) is not None:
        attrs[SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION] = partition

    if (key := kwargs.get("key")) is not None:
        attrs[SpanAttributes.MESSAGING_KAFKA_MESSAGE_KEY] = key

    return attrs

def get_publish_destination_name(
    self,
    kwargs: "AnyDict",
) -> str:
    return cast(str, kwargs["topic"])

def get_consume_attrs_from_message(
    self,
    msg: "StreamMessage[Tuple[ConsumerRecord, ...]]",
) -> "AnyDict":
    raw_message = msg.raw_message[0]

    attrs = {
        SpanAttributes.MESSAGING_SYSTEM: self.messaging_system,
        SpanAttributes.MESSAGING_MESSAGE_ID: msg.message_id,
        SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: msg.correlation_id,
        SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: len(
            bytearray().join(cast(Sequence[bytes], msg.body))
        ),
        SpanAttributes.MESSAGING_BATCH_MESSAGE_COUNT: len(msg.raw_message),
        SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION: raw_message.partition,
        MESSAGING_DESTINATION_PUBLISH_NAME: raw_message.topic,
    }

    return attrs

def get_consume_destination_name(
    self,
    msg: "StreamMessage[Tuple[ConsumerRecord, ...]]",
) -> str:
    return cast(str, msg.raw_message[0].topic)