marshalWriteRequestToRecord converts a PushBytesRequest to a Kafka record.
(partitionID int32, tenantID string, req *tempopb.PushBytesRequest)
| 102 | |
| 103 | // marshalWriteRequestToRecord converts a PushBytesRequest to a Kafka record. |
| 104 | func marshalWriteRequestToRecord(partitionID int32, tenantID string, req *tempopb.PushBytesRequest) (*kgo.Record, error) { |
| 105 | data, err := req.Marshal() |
| 106 | if err != nil { |
| 107 | return nil, fmt.Errorf("failed to marshal record: %w", err) |
| 108 | } |
| 109 | |
| 110 | return &kgo.Record{ |
| 111 | Key: []byte(tenantID), |
| 112 | Value: data, |
| 113 | Partition: partitionID, |
| 114 | }, nil |
| 115 | } |
| 116 | |
| 117 | // Decoder is responsible for decoding Kafka record data back into logproto.Stream format. |
| 118 | // It caches parsed labels for efficiency. |