Feature: Schema Registry support

[ahmeroxa]

Subtasks

• SR: Cached Schema Registry Client #1048
• SR: Extract Avro schema from structured data #1041
• SR: Create Serializer #1042
• SR: Create Deserializer #1043
• SR: Create Processors for Encoding with Schema #1044
• SR: Create Processors for Decoding with Schema #1045

Feature description

Native support for the Confluent Schema Registry API should be added.

This will allow the Kafka connector to translate incoming encoded messages (e.g. Avro) to structured data so that they may be processed by processors as well as be emitted downstream in other formats (e.g. OpenCDC/JSON).

Similarly, the connector will be able to produce data to destination Kafka clusters in encoded formats and register the schema automatically with a provided Schema Registry so that it may be used by downstream consumers.

My request would be to initially ship support for at least Avro but I suspect that supporting the full suite will not be much more effort (Avro, protobuf and JSON Schema).

This would additionally add support for the Apicurio Registry as it exposes a Confluent Schema Registry compatible API.

[lovromazgon]

Should this be a feature specific to the Kafka connector or should this be included in core Conduit? The connector currently reads the message and produces it as raw data to Conduit, I'm thinking that a processor could pick the correct schema from the registry and parse the raw data into structured data.

[ahmeroxa]

I think ideally having the functionality as a processor would be great as it would make it possible to use a Schema Registry with any connector.

[ahmeroxa]

Some useful resources on the Schema Registry:

• High level concept: https://docs.confluent.io/platform/current/schema-registry/fundamentals/index.html#how-it-works
• Subject/Schema naming: https://docs.confluent.io/platform/7.3/schema-registry/fundamentals/serdes-develop/index.html#formats-serializers-and-deserializers
• Go example using Schema Registry: https://github.com/confluentinc/confluent-kafka-go/blob/master/examples/avro_generic_consumer_example/avro_generic_consumer_example.go

[lovromazgon]

Here are some takeaways from our discussion:

• Conduit should provide a processor that can take a schema from a schema registry and parse the raw payload into a structured payload.
• Schemas should be cached in Conduit (we should fetch a schema from the registry only the first time we encounter it).
• The user should be able to configure the schema subject based on record metadata (maybe using a Go template? e.g. my-static-prefix-{{record.Metadata["topic"]}}).
• By default, the processor should use a predefined metadata field as the schema subject (e.g. schema-subject), unless configured otherwise by the user (as mentioned in the previous point). Connectors can populate this metadata field if the resource that they connect provides the info or if there's a convention (e.g. the Kafka connector can follow the same convention as Kafka Connect).
• If a schema is not found the processor should fail.

[gedw99]

This seems very good. Is this on roadmap or stalled ?

[lovromazgon]

It's actively being worked on and should be included in the next release.

[lovromazgon]

Done, documentation is prepared here ConduitIO/conduit-site#37.

[gedw99]

hey,

@lovromazgon @ahmeroxa

Not sure if your interested, but NATS JetStream also has translate support too.
So for example it can translate incoming encoded messages (e.g. Avro), etc.

Code: https://github.com/search?q=repo%3Anats-io%2Fnatscli%20translate&type=code

This shows it off a bit: https://github.com/metatexx/msgcvt that i use.