Help with protobuf decoding by schema registry

[mihelanjelo]

Hello, i have problem with setting up proto schemas for topic.
I have connected schema registry confluentinc/cp-schema-registry:7.0.1
I set schema for key like topic_name-key and for value like topic_name-value
, but still messages are not shown decoded.
I checked schemas many times and decoded messages in code by google lib successfully.
Sometimes i saw error in log like:

org.apache.kafka.common.errors.SerializationException: Error deserializing Protobuf message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

Please, explain by steps how to set it up properly

[Haarolean]

Hi, could you please share an example schema and a message to reproduce this? Thanks.

[Haarolean]

We need a schema and a message example to reproduce this.

[mihelanjelo]

Schema:
Key:


syntax = "proto3";
message EventKey {
int64 some_id = 1;
}
Value :
syntax = "proto3";
message EventValue {
int64 some_id = 1;
Timestamp date = 2;
string some_parent_id = 14;
}
message Timestamp {
int64 seconds = 1;
int32 nanos = 2;
}

Message:
Encoded:
Key: b'\x08\xb9\xdc\xc1\x7f'
Value: b'\x08\xee\xd5\xe5\x8a\x01\x12\x0c\x08\xff\xe5\xdd\x90\x06\x10\xc4\xd9\xb3\xb2\x01Z*\n(\n$13696e0e-4c75-49aa-9c83-0a6faf5074ee\x10\x01r\x0c0_7311196171'

Decoded:
Key:
some_id: 267415097

Value: 
 some_id: 291072750
date {
  seconds: 1645703935
  nanos: 374140100
}
some_parent_id: "0_7311196171"

[mihelanjelo]

I tried to produce massage in kafka-ui and decoding worked, but i need to read decoded messages that are already in topic and encoded in protobuf

[Haarolean]

Thanks for the info, we'll take a look. I'll convert this into an issue.

[mihelanjelo]

Thank you!

[iliax]

Hello @mihelanjelo!
problem is likely in serializer you use when writing to kafka.
In raw bytes message should look like [zero byte, 4 bytes - schema id, remaining payload bytes] see https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#wire-format.

Usually kafka libraries hide first 5 bytes generating for end user. In your case you likely skip 5-bytes prefix, so kafka deserializer can't detect schema for parsing. In such scenario kafka-ui falls back to string deserializing.

See python AvroSerializer usage here: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/avro_producer.py

[mihelanjelo]

Hello. Why avro serializer? I need protobuf serializer and i set schema for protobuf in schema registry.

[mihelanjelo]

i can't understand why i can read and decode protobuf messages in my python code (kafka-python + protobuf libs) but in Kafka-UI it looks like encoded differently and unable to decode (

[iliax]

@mihelanjelo You can read and decode protobuf messages in your code because you know schema of you messages, but kafka-ui does not. To solve such cases schema registry exists.

Schema registry is needed to mark every message you write to kafka with schema id used for serialization. Later when reading messages (by any other consumer) corresponding schema will be extracted from registry and used for deserialzation. Please see concepts described here -> https://docs.confluent.io/platform/current/schema-registry/index.html.

Sorry for pointing you to wrong producing-> here is an example of protobuf producer schemaregistry usage https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/protobuf_producer.py#L76.

From my understanding currently you just serialize your message using protobuf and write to kafka as raw bytes - in that case consumer have no ability to get schema, even it is created in schema-registry.

---

If you want to write raw bytes messages, serialized via protobuf (without using schemaregistry) you can pass kafka-ui topic->protofile mapping. This mapping will be used for messages deserialization by kafka-ui and you will be able to see messages content in ui. (this is not documented, and only values deserialisation supported for such topics). Let me know I will write how to pass such config.

[iliax]

yes, theoretically kafka-ui can get schema by topic name, but it is wrong usage of SR. If for example you will produce message from kafka-ui, message will be generated in proper way (and will be visible), but it likely will break consumers.

[mihelanjelo]

so now all producers and consumers must be connected to SR. But seems possible to create an option to work without those ids in messages as kafka-ui for itself can get ids by topic name...

[iliax]

I think it can lead to SR misusage. We will discuss it think how it can be implemented. Thank you for suggestion.

[messart]

Hi @iliax,

You wrote before
"If you want to write raw bytes messages, serialized via protobuf (without using schemaregistry) you can pass kafka-ui topic->protofile mapping. This mapping will be used for messages deserialization by kafka-ui and you will be able to see messages content in ui. (this is not documented, and only values deserialisation supported for such topics). Let me know I will write how to pass such config."

Could you please describe the way how to apply this mapping? As we have the case that you've described with the topic that contains raw serialized via protobuf messages (without using schemaregistry) and we'd like to view them in kafka-ui in the deserialized way.

Thanks in advance.

[iliax]

Hi @messart, here are things you need to do:

1. mount proto file with messages that are used to kafka-ui docker container (you can paste all messages in one file)
2. set properties for mapping, example:

KAFKA_CLUSTERS_0_PROTOBUF_FILE=/path/to/mounted/protofile.proto
// this proto message will be used for values parsing if there is no specific topic mapping (default)
KAFKA_CLUSTERS_0_PROTOBUF_MESSAGE_NAME=DefaultMessageName
// this proto message will be used for key parsing if there is no specific topic mapping (default)
KAFKA_CLUSTERS_0_PROTOBUF_MESSAGE_NAME_FOR_KEY=DefaultMessageNameForKey

//you can also specify proto message name for specific topic's keys&values:
KAFKA_CLUSTERS_0_PROTOBUF_MESSAGE_NAME_BY_TOPIC_TOPICNAME1=Topic1MessageName
KAFKA_CLUSTERS_0_PROTOBUF_MESSAGE_NAME_FOR_KEY_BY_TOPIC_TOPICNAME1=Topic1KeyMessageName

KAFKA_CLUSTERS_0_PROTOBUF_MESSAGE_NAME_BY_TOPIC_ANOTERTOPICNAME=AonotherTopicMessageName
KAFKA_CLUSTERS_0_PROTOBUF_MESSAGE_NAME_FOR_KEY_BY_TOPIC_ANOTERTOPICNAME=AonotherTopicKeyMessageName

[Haarolean]

Latest docs on serialization