Schema-registry: Using AbstractKafkaAvroSerializer and CachedSchemaRegistryClient together results in a crash

[ksilin]

This issue has been discussed here and here, but with no solution for my case. The schema in our case is generated by the AbstractKafkaAvroSerializer and consumed by the CachedSchemaRegistryClient. For instances of the same (case class) type and same subject, the Serializer generates different schemas, so exactly identityMapCapacity messages can be dispatched to the broker before the RegistryClient fails with java.lang.IllegalStateException: Too many schema objects created for <myTopic>-value.

Since the schema is generated and registered for each message automatically, by two different parts of the schema-registry I see no possibility to prevent the error from happening.

The two workaround options for me are:

• not use schema registry - not a long-term solution obviously
• create a custom SchemaRegistryClient not relying on object identity - doable but looks liek it should be an option provided by the registry as I might create other issues while solving this one ;)

This is the code in question (also using reactive kafka and avro4s):

    val avroProducerSettings: ProducerSettings[String, GenericRecord] = 
      ProducerSettings(system, Serdes.String().serializer(), 
      avroSerde.serializer())
     .withBootstrapServers(settings.bootstrapServer)

    val avroProdFlow: Flow[ProducerMessage.Message[String, GenericRecord, String],
                        ProducerMessage.Result[String, GenericRecord, String],
                        NotUsed] = Producer.flow(avroProducerSettings)

    val avroQueue: SourceQueueWithComplete[Message[String, GenericRecord, String]] = 
      Source.queue(bufferSize, overflowStrategy)
      .via(avroProdFlow)
      .map(logResult)
      .to(Sink.ignore)
      .run()
    ...
    queue.offer(msg)

...

    def toAvro[A](a: A)(implicit recordFormat: RecordFormat[A]): GenericRecord =
      recordFormat.to(a)

    val makeEdgeMessage: (Edge, String) => Message[String, GenericRecord, String] = { (edge, topic) =>
      val edgeAvro: GenericRecord = toAvro(edge)
      val record   = new ProducerRecord[String, GenericRecord](topic, edge.id, edgeAvro)
      ProducerMessage.Message(record, edge.id)
    }

[ksilin]

I have implemented a SchemaRegistryClient not relying on IdentityHashMap. It does not suffer from the memory leak but may have other issues which I am not aware of. Any comments would be appreciated.

Most of the code is adapted from the CachedSchemaRegistryClient and translated to scala.

import java.util

import io.confluent.kafka.schemaregistry.client.rest.entities.{ Config, SchemaString }
import io.confluent.kafka.schemaregistry.client.rest.entities.requests.ConfigUpdateRequest
import io.confluent.kafka.schemaregistry.client.rest.{ RestService, entities }
import io.confluent.kafka.schemaregistry.client.{ SchemaMetadata, SchemaRegistryClient }
import org.apache.avro.Schema

import scala.collection.mutable

class CachingSchemaRegistryClient(val restService: RestService, val identityMapCapacity: Int)
    extends SchemaRegistryClient {

  val schemaCache: mutable.Map[String, mutable.Map[Schema, Integer]] = mutable.Map()
  val idCache: mutable.Map[String, mutable.Map[Integer, Schema]] =
    mutable.Map(null.asInstanceOf[String] -> mutable.Map())
  val versionCache: mutable.Map[String, mutable.Map[Schema, Integer]] = mutable.Map()

  def this(baseUrl: String, identityMapCapacity: Int) {
    this(new RestService(baseUrl), identityMapCapacity)
  }

  def this(baseUrls: util.List[String], identityMapCapacity: Int) {
    this(new RestService(baseUrls), identityMapCapacity)
  }

  def registerAndGetId(subject: String, schema: Schema): Int =
    restService.registerSchema(schema.toString, subject)

  def getSchemaByIdFromRegistry(id: Int): Schema = {
    val restSchema: SchemaString = restService.getId(id)
    (new Schema.Parser).parse(restSchema.getSchemaString)
  }

  def getVersionFromRegistry(subject: String, schema: Schema): Int = {
    val response: entities.Schema = restService.lookUpSubjectVersion(schema.toString, subject)
    response.getVersion.intValue
  }

  override def getVersion(subject: String, schema: Schema): Int = synchronized {
    val schemaVersionMap: mutable.Map[Schema, Integer] =
      versionCache.getOrElseUpdate(subject, mutable.Map())

    val version: Integer = schemaVersionMap.getOrElse(
      schema, {
        if (schemaVersionMap.size >= identityMapCapacity) {
          throw new IllegalStateException(s"Too many schema objects created for $subject!")
        }

        val version = new Integer(getVersionFromRegistry(subject, schema))
        schemaVersionMap.put(schema, version)
        version
      }
    )
    version.intValue()
  }

  override def getAllSubjects: util.List[String] = restService.getAllSubjects()

  override def getByID(id: Int): Schema = synchronized { getBySubjectAndID(null, id) }

  override def getBySubjectAndID(subject: String, id: Int): Schema = synchronized {
    val idSchemaMap: mutable.Map[Integer, Schema] = idCache.getOrElseUpdate(subject, mutable.Map())
    idSchemaMap.getOrElseUpdate(id, getSchemaByIdFromRegistry(id))
  }

  override def getSchemaMetadata(subject: String, version: Int): SchemaMetadata = {
    val response = restService.getVersion(subject, version)
    val id       = response.getId.intValue
    val schema   = response.getSchema
    new SchemaMetadata(id, version, schema)
  }

  override def getLatestSchemaMetadata(subject: String): SchemaMetadata = synchronized {
    val response = restService.getLatestVersion(subject)
    val id       = response.getId.intValue
    val version  = response.getVersion.intValue
    val schema   = response.getSchema
    new SchemaMetadata(id, version, schema)
  }

  override def updateCompatibility(subject: String, compatibility: String): String = {
    val response: ConfigUpdateRequest = restService.updateCompatibility(compatibility, subject)
    response.getCompatibilityLevel
  }

  override def getCompatibility(subject: String): String = {
    val response: Config = restService.getConfig(subject)
    response.getCompatibilityLevel
  }

  override def testCompatibility(subject: String, schema: Schema): Boolean =
    restService.testCompatibility(schema.toString(), subject, "latest")

  override def register(subject: String, schema: Schema): Int = synchronized {
    val schemaIdMap: mutable.Map[Schema, Integer] =
      schemaCache.getOrElseUpdate(subject, mutable.Map())

    val id = schemaIdMap.getOrElse(
      schema, {
        if (schemaIdMap.size >= identityMapCapacity)
          throw new IllegalStateException(s"Too many schema objects created for $subject!")
        val id: Integer = new Integer(registerAndGetId(subject, schema))
        schemaIdMap.put(schema, id)
        idCache(null).put(id, schema)
        id
      }
    )
    id.intValue()
  }
}

[yarshad]

Hi, Thanks for sharing your solution. I am experiencing the same issue, and your solution helped.

I am curious to know why this is still a bug, and how all the companies using KAFKA - have either not encountered this issue or if they did why its still unresolved.

[ksilin]

Glad I was able to help someone, @yarshad I too am wondering why there is no discussion, looks like we are the only one with the issue here 🙈 🙉 🙊

[mageshn]

@ksilin the best solution is to ensure same schema object is generated.reused across avro records. Even if you override the client, it would be expensive to compare the complete canonical schema for every produce request. The best is to fix the source to generate schema object.

[mageshn]

@yarshad and @ksilin In most cases, when pre compiled SpecificAvro class is used this doesn't happen and that's the reason its not really an issue for most people. Also, it is recommended to follow that pattern so that your application also doesn't go through the GC overhead.

[yarshad]

@mageshn - Does this mean I will not get this exception if I am publishing SpecificRecord to Kafka. Currently I am publishing generic records.

Thanks

[mageshn]

@yarshad you should give that a try. It should likely fix your issues

[ksilin]

@mageshn @yarshad thank you for the help :) unfortunately, my attempts at using SpecificRecord failed because of the way AbstractKafkaAvroDeserializer generates the reader schema - it assumes the existence of a no-arg ctor for the msg class, which is not sound for scala case classes. This issue has been described here: https://github.com/confluentinc/schema-registry/issues/462 but has not been addressed yet.

Is there a way around this?

[ksilin]

I have tried to circumvent the issue by providing a static schema to an overloaded deserializer. However the core issue of expecting no-arg ctors for message instances seems to occur in multiple places.

Caused by: java.lang.RuntimeException: java.lang.NoSuchMethodException: com.example.avro.SimpleMessage.<init>()
    at org.apache.avro.specific.SpecificData.newInstance(SpecificData.java:353)
    at org.apache.avro.specific.SpecificData.newRecord(SpecificData.java:369)
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:212)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:175)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:153)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:145)
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:133)

Since the msg is a scala case class, it does not have a no-arg ctor.

[heliocentrist]

@ksilin I think the problem might be in this piece of code:

def toAvro[A](a: A)(implicit recordFormat: RecordFormat[A]): GenericRecord =
      recordFormat.to(a)

Do you create an implicit instance of RecordFormat, or do you let avro4s do it for you? If you don't create your own implicit instance, then I believe avro4s will create a new one every time you call this method, and this will lead to a new schema instance being generated every time you serialize something. I would suggest creating an implicit instance of RecordFormat[Edge] and trying to use the standard client and serializer again.

[ksilin]

@heliocentrist nice! Many thanks. This seems to solve the issue for me. Could you please elaborate how the RecordFormat influences the schema caching here? I assumed, the GenericRecord instance does not return the format/schema it has been generated with but the schema is derived by the AbstractKafkaAvroSeriliazer. Your example implies otherwise.

[heliocentrist]

@ksilin Happy to help! First, I should say that I'm not an expert in avro4s, so I might misunderstand some details. The reference to the schema used to generate the record is stored in the GenericRecord instance, so when the serializer calls record.getSchema it receives that reference. If the reference is the same in all GenericRecords generated with the same schema, then you don't have the problem. But if a new schema object is created every time a GenericRecord is created, then of course you have the problem with the identity map.
Without getting into too much detail, the RecordFormat instance for a type T stores the schema instance for this type. If you create the implicit instance of RecordFormat[T], then every time you call your toAvro method this same instance is used, and it follows that the same instance of the schema is injected into the GenericRecords. If you don't have an implicit RecordFormat[T] defined, then avro4s will create it for you, but it will create a new one every time you call the method. Consequently, a new schema instance will be generated and injected every time you create a GenericRecord.
Hope it makes sense!

[ksilin]

@heliocentrist Makes perfect sense. Thanks again!

[ghost]

@ksilin i have one doubt regard to implementing custom CachedSchemaRegistryClient, how you would be referring this custom implemented SchemaRegistryClient in your code?

[ksilin]

@shivashankar-awanti I use it in the serializer/deserializer in order for it to be passed to the AbstractKafkaAvroSerializer

`mport java.util.Collections

import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
import org.apache.avro.Schema
import org.apache.avro.generic.GenericRecord
import org.apache.kafka.common.serialization.{ Deserializer, Serde, Serdes, Serializer }

class GenericAvroSerde(serde: Serde[GenericRecord]) extends Serde[GenericRecord] {

def this(client: SchemaRegistryClient,
props: java.util.Map[String, _],
schema: Option[Schema] = None) =
this(
Serdes.serdeFrom(new GenericAvroSerializer(client, schema),
new GenericAvroDeserializer(client, props))
)

def this(client: SchemaRegistryClient) = this(client, Collections.emptyMap[String, AnyRef])

def this() = this(Serdes.serdeFrom(new GenericAvroSerializer, new GenericAvroDeserializer))

def serializer: Serializer[GenericRecord] = serde.serializer

def deserializer: Deserializer[GenericRecord] = serde.deserializer

def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = {
serde.serializer.configure(configs, isKey)
serde.deserializer.configure(configs, isKey)
}

def close(): Unit = {
serde.serializer.close()
serde.deserializer.close()
}
}`

[ghost]

@ksilin Thanks for the response. Im approaching the same way, please find the below code snippet where im passing message to convert it to GenericAvroSerde:

I'm not getting where to make changes exactly to avoid the error "Caused by: java.lang.IllegalStateException: Too many schema objects created for AnalyticThermostatMessagSchema-value! at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:152) ~[kafka-schema-registry-client-4.1.0.jar!/:na]"

i have also checked the class CachedSchemaRegistryClient, i'm not getting what exactly it needs to be changed.

package io.confluent.kafka.streams.serdes.avro;

import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
import java.util.Map;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.common.annotation.InterfaceStability.Unstable;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;

@Unstable
public class GenericAvroSerde implements Serde {
private final Serde inner;

public GenericAvroSerde() {
    this.inner = Serdes.serdeFrom(new GenericAvroSerializer(), new GenericAvroDeserializer());
}

GenericAvroSerde(SchemaRegistryClient client) {
    if (client == null) {
        throw new IllegalArgumentException("schema registry client must not be null");
    } else {
        this.inner = Serdes.serdeFrom(new GenericAvroSerializer(client), new GenericAvroDeserializer(client));
    }
}

public Serializer<GenericRecord> serializer() {
    return this.inner.serializer();
}

public Deserializer<GenericRecord> deserializer() {
    return this.inner.deserializer();
}

public void configure(Map<String, ?> serdeConfig, boolean isSerdeForRecordKeys) {
    this.inner.serializer().configure(serdeConfig, isSerdeForRecordKeys);
    this.inner.deserializer().configure(serdeConfig, isSerdeForRecordKeys);
}

public void close() {
    this.inner.serializer().close();
    this.inner.deserializer().close();
}

}