Protobuf Key and Header in Kafka Source

[seilerre]

We are currently using a Kafka Source which reads in protobuf like so

      kafka_metrics:
        type: "kafka"
        bootstrap_servers: "bootstrap.strimzi.svc.cluster.local:9093"
        group_id: "metrics.prod"
        topics:
          - "^edge.*metrics.*"
        sasl:
          enabled: true
          mechanism: SCRAM-SHA-512
          username: ${SRC_KAFKA_SASL_USERNAME}
          password: ${SRC_KAFKA_SASL_PASSWORD}
        decoding:
          codec: protobuf
          protobuf:
            desc_file: /opt/proto-store/streamingevent.desc
            message_type: edge.StreamingProtoEventV1Value

We now also want to encode the key and header in protobuf before sending to Kafka. Will Vector be able to read this? How can I include this?

[pront]

Hi @seilerre,

We now also want to encode the key and header in protobuf before sending to Kafka.

Vector will be able to decode everything based on the description file and message type you specify. So if you change your protos to include "key and header" then those will be part of the decoded Vector event. Let me know if I am missing something.

[seilerre]

@pront : Thanks for your reply! I mean, we can only specify message_type. So if message_type is an object, what do we have to put in there so Vector recognizes key and value fields?

message_type: StreamingProtoEventV1

message StreamingProtoEventV1 {
  optional StreamingProtoEventV1Key key = 1;
  optional StreamingProtoEventV1Value value = 2;
}
 
message StreamingProtoEventV1Key {
  optional string someKey = 1;
}
 
message StreamingProtoEventV1Value {
  optional string someKey = 1;

}

so for example in AKHQ (https://akhq.io/docs/configuration/protobuf.html) you can set a key and value proto file:

key-message-type: "org.akhq.utils.Row"
value-message-type: "org.akhq.utils.Envelope"

[seilerre]

@pront Didn't have time to experiment with this but my assumption would be that this doesnt work. And if it does, we would send the key over the wire twice right, once in the Kafka key field, once as part of the message. I am thinking about https://vector.dev/docs/reference/vrl/functions/#parse_proto. Maybe we could do something like: .key_struct = parse_proto!(.message_key, "/proto/desc.pb", "mypkg.KeyMsg"). I will try this out in the next days and maybe mark this as an answer. My only concern: Vector expects the Kafka Key and put it per default into message_key. Will this fail before I can reach a transform?

[seilerre]

This worked great by doing:

decoding:
          codec: bytes

and then

    transforms:
      parse_proto:
        type: remap
        inputs:
          - kafka
        source: |-
          .parsed_value = parse_proto!(.message, desc_file: "/opt/proto-store/proto.desc", message_type: "mes.Value")
          .parsed_key = parse_proto!(.message_key, desc_file: "/opt/proto-store/proto.desc", message_type: "mes.Value")
          del(.message)
          del(.message_key)