MockSchemaRegistryClient now throws IllegalStateException: Invalid schema type JSON

[Zordid]

Coming from version 8.1.0 and updating to 8.1.1, there is a new bug introduced in the MockSchemaRegistryClient:

the old implementation ALWAYS worked with Json schemas, even if the providers were not initialized:

val mockClient = MockSchemaRegistryClient()

just worked when using this in conjunction with a KafkaJsonSchemaSerializer.

Updating to 8.1.1 leads to an exception:

org.apache.kafka.common.errors.SerializationException: Error serializing JSON message
	at io.confluent.kafka.serializers.json.AbstractKafkaJsonSchemaSerializer.serializeImpl(AbstractKafkaJsonSchemaSerializer.java:188)
	at io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer.serialize(KafkaJsonSchemaSerializer.java:115)
	at io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer.serialize(KafkaJsonSchemaSerializer.java:90)
Caused by: com.google.common.util.concurrent.UncheckedExecutionException: java.lang.IllegalStateException: Invalid schema type JSON
	at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2085)
	at com.google.common.cache.LocalCache.get(LocalCache.java:4011)
	at com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:4034)
	at com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:5010)

The reason for this comes from this piece of code that was changed recently:

  @Override
  public Optional<ParsedSchema> parseSchema(Schema schema) {
    try {
      return Optional.of(parsedSchemaCache.get(schema));
    } catch (ExecutionException e) {
      return Optional.empty();
    }
  }

Here, only exceptions of type ExecutionException are handled, but others are ignored and surface!

This is why the KafkaJsonSchemaSerializer cannot do it's auto-registering anymore:

    if (autoRegisterSchema) {
        restClientErrorMsg = "Error registering JSON schema: ";
        io.confluent.kafka.schemaregistry.client.rest.entities.Schema s =
            registerWithResponse(subject, schema, normalizeSchema, propagateSchemaTags);
        if (s.getSchema() != null) {
          Optional<ParsedSchema> optSchema = schemaRegistry.parseSchema(s);
          if (optSchema.isPresent()) {
            schema = (JsonSchema) optSchema.get();
            schema = schema.copy(s.getVersion());
          }
        }
        schemaId = new SchemaId(JsonSchema.TYPE, s.getId(), s.getGuid());
      } 

I am horrified on what kind of changes are done without any unit tests... :-(

[rayokota]

Thanks for digging into this. I apologize for breaking your tests.

This stems from a change intended to make MockSchemaRegistryClient behave exactly like CachedSchemaRegistryClient. We found that the mock SR client was previously too lenient; it would succeed in scenarios where the real client (without a specific schema prrovider) would throw an error. We wanted to close that gap so tests would be more representative of production behavior. The fix to the mock SR client is a copy of the code from CachedSchemaRegistryClient:

  @Override
  public Optional<ParsedSchema> parseSchema(Schema schema) {
    try {
      return Optional.of(parsedSchemaCache.get(schema));
    } catch (ExecutionException e) {
      return Optional.empty();
    }
  }

To fix your tests, you just need to initialize the mock with the JSON provider explicitly:

new MockSchemaRegistryClient(ImmutableList.of(new JsonSchemaProvider()));

I hope this helps!