Setup Avro schemas

[inakianduaga]

Setup Avro schemas

[inakianduaga]

Notes on schema registry integration

• Official docs on serialization/deserialization

• http://www.programcreek.com/java-api-examples/index.php?api=io.confluent.kafka.serializers.KafkaAvroSerializer

• http://docs.confluent.io/3.0.0/schema-registry/docs/serializer-formatter.html

• We might try without using anything special and try to get the schema registry working as simple as possible with akka kafka streams

Avro producer example

Example of Akka stream avro record production (without using case classes)

Here is a snippet of my scala code that failed:

//My simple avro schema
val key = "key1"
  val f1 = "str1"
  val f2 = "str2"
  val f3 = "int1"
val userSchema = """
    {
        "fields": [
            { "name": "str1", "type": "string" },
            { "name": "str2", "type": "string" },
            { "name": "int1", "type": "int" }
        ],
        "name": "myrecord",
        "type": "record"
    }"""

  val parser = new Schema.Parser
  val schema = parser.parse(userSchema)

//My SchemaReg client
private val avroRecord: GenericData.Record = new GenericData.Record(schema)
val client = new CachedSchemaRegistryClient("http://hostname:8181",1000)

//My kafka producer settings with Akka Stream Kafka
val producerSettings = 
          ProducerSettings(context.system,
              new io.confluent.kafka.serializers.KafkaAvroSerializer(client),
              new io.confluent.kafka.serializers.KafkaAvroSerializer(client))
          .withBootstrapServers("sherpavm:6667")
          .withProperty("schema.registry.url", "http://sherpavm:8181")

//My producer function with with Akka Stream Kafka
def testProducer(epoch: Int) = Source(epoch to epoch+20)
          .map { i =>
            addRecords(avroRecord, s"st1-1-${i}", s"st1-2-${i}", i)
            new ProducerRecord[Object, Object](topic, key, avroRecord)
          }
          .runWith(Producer.plainSink(producerSettings))

def addRecords(avroRecord: GenericData.Record, str1: String, str2: String, int1: Int) = {
    if (str1 != null) avroRecord.put(f1, str1)
    if (str2 != null) avroRecord.put(f2, str2)
    avroRecord.put(f3, int1)
  }

[inakianduaga]

Problems:

Not sure how to handle reading with custom readers (projections)

https://github.com/confluentinc/schema-registry/blob/master/avro-serializer/src/main/java/io/confluent/kafka/serializers/AbstractKafkaAvroDeserializer.java#L99

Funcionality is provided, but not sure how to get it to work automatically when passing the decoder. We might have to extend the KafkaAvroDeserializer with a reader Schema that if passed will be used to deserialize, otherwise defaults to standard

• This pull request is the one that adds support for specifying a separate class I think
  https://github.com/confluentinc/schema-registry/pull/274/files

• Another option: Use bytearray deserializer and run the deserialization manually using the decoder schema method, instead of trying to do it automatically

[inakianduaga]

Schema evolution rules

https://docs.oracle.com/cd/E26161_02/html/GettingStartedGuide/schemaevolution.html

Aliases:

Seems schema registry doesn't accept aliases as being backwards compatible:

https://groups.google.com/forum/#!topic/confluent-platform/0NlrxFD5FHk
confluentinc/schema-registry#453

[inakianduaga]

Example of writing our own deserializer

http://stackoverflow.com/questions/36697350/avro-with-kafka-deserializing-with-changing-schema
https://blog.knoldus.com/2016/08/02/scala-kafka-avro-producing-and-consuming-avro-messages/

which we might use to have a GenericRecord instead of an IndexedRecord or something like that

[inakianduaga]

Compilation error bug

Seems the class ImageRequestDeserializer on the UI backend is breaking compilationn for some reason, we need to remove it and investigate
Update: Running sbt last compile yields

[error] /var/kafka-playground/ui-backend/app/services/Kafka.scala:51: too many arguments for constructor ImageProcessedDeserializer: ()deserializers.ImageProcessedDeserializer
[error]       val avroDeserializer = new ImageProcessedDeserializer(schemaRegistryClient)