[exporter/kafka] Enable Trace batch chunking before exporting to kafka

[shivanshuraj1333]

Related to: #36982

Currently kafka exporter is not exporting messaging by adhering to the limits set by MaxMessageBytes, which can cause trace packets to drop when exceeding the limit.

This PR aims to fix that by effectively dividing the spans into chunks which have size under the limit set by MaxMessageBytes, and then individually send these chunks.

[shivanshuraj1333]

This is almost done, would open it up for review after manually testing it with kafka.

[yurishkuro]

this can be checked in Config.Validate method once

[yurishkuro]

this is still trying to marshal one large trace as a single message, it's not going to solve the problem in the issue.

[shivanshuraj1333]

this is used to prepare the msg at line 151, at 159 I'm checking the packet size, and then doing the splitting.
AFAIU, marshaling can happen at bigger trace, but the export has to be chunked, right?

[yurishkuro]

you cannot break up already marshaled bytes, it creates invalid payload that cannot be parsed by the consumer. This would break the data contract for the topic.

[shivanshuraj1333]

I think there's some confusion, if I am not wrong.

So, the logic that I am trying to implement is,

• for Jaeger, I'm adding spans to the chunk while checking the size (easy). (since the spans already have resource info)

For the pdata_marshaler.go and marshaler.go I intend to do the following.

• get the bytes by marshaling the whole trace in a given batch
  • if the size is under limits, append that to a chunk (*sarama.ProducerMessage)
  • if the size exceeds, split the spans into two halves (with the same resource) for the current batch
    • do this binary search till the size of new split spans after marshaling comes out to be under limits
  • repeat this for all the batches

I have updated the Marshaller here which contains []*sarama.ProducerMessage

Here I'm marshaling the first half and here marshaling the second half, which are the two halves of the original trace.

The above logic ensures that every slice is under the limits. []*ProducerMessageChunks contains -> []*sarama.ProducerMessage

and each value in []*sarama.ProducerMessage is a valid split of initial trace, and then the marshaled value, not the splitting of bytes.

Kindly correct me if I am missing something. (maybe I need to fix some things in the code, but do you agree the above logic works?)

[shivanshuraj1333]

@yurishkuro when you get time, can you review my approach?

tldr: I am not appending bytes randomly, rather I am ensuring that the marshled trace when converted to *sarama.ProducerMessage follows the size, if it doesn't, I split the trace into two traces and then marshal again.

[yurishkuro]

why does this type need to change? The exporter is getting an array of messages and calls SendMessages(msgs []*ProducerMessage) on sarama.Producer. At that point, is the max message size applied to the whole batch or to each individual message? Intuitively it's the latter (I don't see evidence to the contrary), and since each message contains exactly one span per the previous logic, there's really nothing else to do here.

And if my assumption is correct, I don't think the introduction of ProducerMessageChunks struct is necessary - just make sure that each message is not larger than max.