Integration with Confluent Schema Registry?

[jacopofar]

Hello,
in my company we developed a sort of wrapper around this library in order to integrate the Confluent schema registry and perform the Avro serialization. We are now considering open sourcing it, would you consider integrating it in this library (as opposed to a separate package)? It would add requests and fastavro as extra dependencies.

[jacopofar]

For the interested, this is the library: https://github.com/flix-tech/kafka-schema-registry

[FrsECM]

@jacopofar, thanks for the hint !

Just a question is there a way to get the schema from consumer side ?

[jacopofar]

@FrsECM yes, same logic. You read the message and the first bytes are the ID of the schema on the registry:

from fastavro import schemaless_reader
import requests

@lru_cache()
def get_schema(schema_id: Any) -> Dict:
    url = SCHEMA_REGISTRY_URL + '/schemas/ids/' + str(schema_id)
    response = requests.request("GET", url)
    return json.loads(json.loads(response.text)['schema'])

_, schema_id = struct.unpack('>bI', msg[:5])
  schema = get_schema(schema_id)
  data = schemaless_reader(
      io.BytesIO(msg[5:]),
      schema,
      return_record_name=return_record_name,
  )

when publishing you can send the JSON to the schema registry and it automatically gives you its id if it's already published so I don't have to worrry about getting it first