Client method getSchemas() does not inform about the version of the returned AVRO schemas

[fmiguelez]

Even though REST call to endpoint /schemas returns corresponding version with each schema entry, in the object ParsedSchema (implemented by AvroSchema) is set to null.

The BUG is in this line of AvroSchemaProvider:

https://github.com/confluentinc/schema-registry/blob/9e7ab57e7f1d5664bc08f42ad9678c5b68a769d8/client/src/main/java/io/confluent/kafka/schemaregistry/avro/AvroSchemaProvider.java#L55C11-L55C15

It should set the version that comes in Schema object:

  @Override
  public ParsedSchema parseSchemaOrElseThrow(Schema schema, boolean isNew, boolean normalize) {
    try {
      return new AvroSchema(
          schema.getSchema(),
          schema.getReferences(),
          resolveReferences(schema),
          schema.getMetadata(),
          schema.getRuleSet(),
          schema.getVersion(), // Proposed change
          (validateDefaults || normalize) && isNew
      );
    } catch (Exception e) {
      log.error("Could not parse Avro schema", e);
      throw e;
    }
  }

[filpano]

Hunting through the commit history, this seems to be the earliest commit where the AvroSchemaProvider class was created. The version identifier has always been hardcoded to be null for some reason.

If you are using the Schema Registry Client directly, there is a rather trivial workaround in that one can extend AvroSchemaProvider#parseSchemaOrElseThrow and include this information from the fetched io.confluent.kafka.schemaregistry.client.rest.entities.Schema and then pass this to SchemaRegistryClientFactory.newClient.

public class VersionedAvroSchemaProvider extends AvroSchemaProvider {

    private boolean validateDefaults = false;

    @Override
    public void configure(Map<String, ?> configs) {
        super.configure(configs);
        String validate = (String) configs.get(AVRO_VALIDATE_DEFAULTS);
        validateDefaults = Boolean.parseBoolean(validate);
    }

    @Override
    public ParsedSchema parseSchemaOrElseThrow(Schema schema, boolean isNew, boolean normalize) {
        Assert.state(schema.getVersion() != null, "Failed to extract version id from schema while fetching schema");
        try {
            return new AvroSchema(
                    schema.getSchema(),
                    schema.getReferences(),
                    resolveReferences(schema),
                    schema.getMetadata(),
                    schema.getRuleSet(),
                    schema.getVersion(),
                    (validateDefaults || normalize) && isNew
            );
        } catch (Exception e) {
            log.error("Could not parse Avro schema", e);
            throw e;
        }
    }
}

(it's also rather unfortunate that the validateDefaults property is private, as this shows)

FWIW I agree that this is a bug.