Aiven-Open / Aiven-Open/karapace
Schema normalization issue - Protobuf errors and incompatibility with Confluent Schema Registry
- Vorherrschende Sprache
- Python
- Sterne
- 634
- Forks
- 110
- Ø Merge
- 4 T. 7 Std.
- Gemergte PRs (30 T.)
- 4
Beschreibung
# What happened?
Hey team 👋🏻
I think we just found a bug (or need some clarification?) with Protobuf in Karapace (incompatibility with Confluent SR):
We are using an in-house Python client (wrapper of [confluent_kafka](https://github.com/confluentinc/confluent-kafka-python)), and during e2e testing against Dev cluster on Aiven, we faced the following serialization error during consumption:
```
2025-03-21 17:25:51,828 - INFO - connect - [IPv4 ('127.0.0.1', 55797)]>: connecting to 127.0.0.1:55797 [('127.0.0.1', 55797) IPv4]
2025-03-21 17:25:51,873 - INFO - _send_single_request - HTTP Request: GET http://localhost:8085/subjects/testing-template-examples-topic1-test-v1-value/versions/latest?format=serialized "HTTP/1.1 200 OK"
Exception in thread Thread-1 (run_protobuf_producer_sync):
Traceback (most recent call last):
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/confluent_kafka/schema_registry/protobuf.py", line 179, in _str_to_proto
file_descriptor_proto.ParseFromString(serialized_pb)
google.protobuf.message.DecodeError: Error parsing message with type 'google.protobuf.FileDescriptorProto'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/threading.py", line 1045, in _bootstrap_inner
self.run()
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/threading.py", line 982, in run
self._target(*self._args, **self._kwargs)
File "/Users/eladleev/github/myPythonClientExample/main.py", line 140, in run_protobuf_producer_sync
protobuf_producer.produce_sync(
File "/Users/eladleev/Library/Python/3.11/lib/python/site-packages/myPythonClient/clients/protobuf_producer.py", line 100, in produce_sync
else self._value_serializer(
^^^^^^^^^^^^^^^^^^^^^^^
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/confluent_kafka/schema_registry/protobuf.py", line 582, in __call__
fd_proto, pool = self._get_parsed_schema(latest_schema.schema)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/confluent_kafka/schema_registry/protobuf.py", line 611, in _get_parsed_schema
fd_proto = _str_to_proto("default", schema.schema_str)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/confluent_kafka/schema_registry/protobuf.py", line 182, in _str_to_proto
raise SerializationError(str(e))
confluent_kafka.serialization.SerializationError: Error parsing message with type 'google.protobuf.FileDescriptorProto'
```
I believe it related to the way Karapace handle [Schema Normalization](https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#schema-normalization) with Non-qualified names vs. fully-qualified names.
Our schema:
```
syntax = "proto3";
package tech.elad.testing.template_examples.v1;
import "google/protobuf/timestamp.proto";
message Topic1Message {
int32 MyID = 1;
string correlationID = 2;
int32 profileID = 3;
google.protobuf.Timestamp cutOffDateTime = 4;
bool canary = 5;
google.protobuf.Timestamp actionedAt = 6;
string actionedBy = 7;
string newField = 8;
bool newBool = 9;
bool anotherNewBool = 10;
bool anotherField = 11;
bool yetAnotherField = 12;
}
```
We register all schemas using `normalize=true`.
For testing purposes, we used Confluent Schema Registry - any everything work as expected.
When we let the client register the schemas using `"auto.register.schemas": True`, we get the following schema, which work well:
```
syntax = "proto3";
package tech.elad.testing.template_examples.v1;
import "google/protobuf/timestamp.proto";
message Topic1Message {
int32 MyID = 1;
string correlationID = 2;
int32 profileID = 3;
.google.protobuf.Timestamp cutOffDateTime = 4;
bool canary = 5;
.google.protobuf.Timestamp actionedAt = 6;
string actionedBy = 7;
string newField = 8;
bool newBool = 9;
bool anotherNewBool = 10;
bool anotherField = 11;
bool yetAnotherField = 12;
}
```
(_note the dot with_ `google.protobuf.Timestamp`)
Could you advise please?
# What did you expect to happen?
Consumption should work well given that set register the schema with `normalize=true` and consumed it with `normalize.schemas=true`
# What else do we need to know?
The issue was raised with our Aiven support team, and opening this issue upon request
Beitragsleitfaden
Bewertung
Dieses Issue wurde noch nicht bewertet.