Is it possible to reference an external schema from inside multiple *.avsc files?

[undrivendev]

Hello,
thanks for this project.

We need to upload our schemas to the Confluent Schema Registry.

The functionality of adding references to external schemas seems to be supported by the SR and by AVRO itself, is there any way we can write our AVRO schemas, including external references, and generate C# classes from them?

Something like this:

Parent schema:

{
    "schema": {
        "type": "Test.Namespace.ReferencedType",
        "name": "MainType",
        "namespace": "Test.Namespace"
    },
    "schemaType": "AVRO",
    "references": [
        {
            "name": "Test.Namespace.ReferencedType",
            "subject": "Test.Namespace.ReferencedType",
            "version": 1
        }
    ]
}

Referenced schema:

{
  "type": "record",
  "namespace": "Test.Namespace",
  "name": "ReferencedType",
  "fields": [
     ...
  ]
}

Sources:

• Add support for schema pluggability and schema references confluentinc/schema-registry#1280
• Example usage: confluentinc/cp-schema-registry:5.5.0: Avro schema reference throws InvalidSchemaException confluentinc/schema-registry#1439

[dstelljes]

This isn’t something that Chr.Avro.Confluent supports, but we probably should. Currently, the serdes only use SchemaString (https://github.com/ch-robinson/dotnet-avro/blob/9.5.1/src/Chr.Avro.Confluent/Confluent/AsyncSchemaRegistryDeserializer.cs#L196). To make references work, I think we would only have to fetch all references and parse them into JsonSchemaReaderContext ahead of reading the schema itself.

[mattlangis]

Hello @dstelljes ,

Love your work!

Exactly as you are suggesting, I was able to introduce a new IJsonSchemaReaderCase that could fetch from schema registry if a schema is split. I think the same type of work could be done on a file version.

To make it more flexible, I think it would be great if the SchemaReader.Read(registration.SchemaString); would be async, it would allow to expend even more. The tricky part I guess is the IEnumerable<Func<>> ... Anyhow, I was wonder if you would like that we (I or you) open an issue and I could contribute on that side. I have a JsonSchemaRegistrySchemaReaderCase done already (but had to use .GetAwaiter().GetResult(), which I'm by far not a fan).

Let me know if you want to discuss this further :)

[dstelljes]

Yeah, I’d be happy to take a look at your JsonSchemaRegistrySchemaReaderCase implementation if you’re willing to share. I’m also curious what you have in mind for a “file version”; is there a reference format you’re thinking of?

[mattlangis]

Hi,

Sorry for late reply, required permissions from the company first :)

Here is what the JsonSchemaRegistrySchemaReaderCase looks like:

internal class JsonSchemaRegistrySchemaReaderCase : IJsonSchemaReaderCase
{
    private readonly ISchemaRegistryClient _schemaRegistryClient;
    private readonly IJsonSchemaReader _reader;

    public JsonSchemaRegistrySchemaReaderCase(ISchemaRegistryClient schemaRegistryClient, IJsonSchemaReader reader)
    {
        _schemaRegistryClient = schemaRegistryClient;
        _reader = reader;
    }

    public JsonSchemaReaderCaseResult Read(JsonElement element, JsonSchemaReaderContext context)
    {
        if (element.ValueKind != JsonValueKind.String)
            return JsonSchemaReaderCaseResult.FromException(new UnknownSchemaException($"{nameof(JsonSchemaReaderCaseResult)} can only be applied to named schema references."));

        var name = element.GetString();

        if (string.IsNullOrEmpty(name))
        {
            return JsonSchemaReaderCaseResult.FromException(new UnknownSchemaException($"Empty schema name is invalid"));
        }

        // Since the Reads are not async, we need to block here.
        var registeredSchema = _schemaRegistryClient.GetLatestSchemaAsync(name).GetAwaiter().GetResult();
        if (registeredSchema == null)
        {
            return JsonSchemaReaderCaseResult.FromException(new UnknownSchemaException($"\"{name}\" is not a registered schema."));
        }

        var schema = _reader.Read(registeredSchema.SchemaString, context);

        return JsonSchemaReaderCaseResult.FromSchema(schema);
    }
}

It's very simple: since it's the last JsonSchemaReaderCase, it means none before worked/didn't found anything. At this point, it go to the schema registry and ask for the schema. Once it has the JSON, it will pass it to the IJsonSchemaReader (in a recursive way) until everything is properly loaded in the JsonSchemaReaderContext.

Very basic implementation, but so far works pretty well. What I dislike is by far the .GetAwaiter().GetResult(), that's because the IJsonSchemaReaderCase.Read doesn't have any async version.

For the "file version", it's hard to make it super robust, but if we take in consideration that avrogen generate at the specified namespace, then maybe there are way to find the class define in the type. Or at least try to load it?
That is definitively to be investigated.

[dstelljes]

Awesome, thanks for sharing! Apologies for the late response on my part.

One thing that worries me a little about this approach is the GetLatestSchemaAsync. References point to a specific version so there could be subtle correctness issues introduced by assuming latest. I suspect that we’d have to recursively fetch and cache references in AsyncSchemaRegistryDeserializer since that’s the layer where we have the full Registry response.

then maybe there are way to find the class define in the type

Just to clarify, are you talking about the C#-to-Avro direction or the Avro-to-C# direction? I think there are opportunities to extend both dotnet avro create and dotnet avro generate to do more intelligent splitting.