Question: avro plus Schema Registry? about kafka-streams-scala HOT 9 CLOSED

In kafka-streams-query we had schema registry working (https://github.com/lightbend/kafka-streams-query/blob/develop/examples/example-dsl/src/main/scala/com/lightbend/kafka/scala/iq/example/WeblogProcessing.scala#L92-L94). In fact we had the option of switching it on through configuration. And it worked. But we did not try with case classes there. We need to do some sanity checks - but I think it should not be difficult.

from kafka-streams-scala.

Thanks for the reply, so I am looking at the example and I see the configuration being passed in obviously but I notice that that project does not depend on io.confluent.kafka-streams-avro-serde which as far as I can tell is where the 'schema-registry aware' serdes live so I wonder if that example would actually end up hitting the schema-registry if there was one?

from kafka-streams-scala.

Isn't this the one ? https://github.com/lightbend/kafka-streams-query/blob/develop/examples/example-dsl/src/main/scala/com/lightbend/kafka/scala/iq/example/WeblogProcessing.scala#L12

from kafka-streams-scala.

that is the configuration property but I think that property gets ignored unless you use one of these https://github.com/confluentinc/schema-registry/tree/master/avro-serde/src/main/java/io/confluent/kafka/streams/serdes/avro . which get included when your project depends on io.confluent.kafka-streams-avro-serde. I could be wrong but thats how I got it working

I have got a messy version of this projects StreamToTableJoinScalaIntegrationTestImplicitSerdes set up hitting my schema-registry working (to an extent but think I am close) using all GenericAvroSerde and avro4s to convert back and forth from GenericRecord to case classes. The only thing stopping me from being able to use SpecificAvroSerde is being able to generate an instance of SpecificRecord for my case classes as avro4s does not support that it seems.

if it would help at all to see what I have done I could clean it up and push to a branch here

from kafka-streams-scala.

Since I was trying to abstract over the optionality of schema registry, I kept the serializers here .. https://github.com/lightbend/kafka-streams-query/tree/develop/examples/example-dsl/src/main/scala/com/lightbend/kafka/scala/iq/example/serializers and use the AppSerializers as the mixin .. The implementation is a bit old and possibly can be cleaned though. HTH.

from kafka-streams-scala.

ahhhhh thank-you, let me have a look

from kafka-streams-scala.

1. feel free to tell me this isnt the place for this convo, dont want to clutter up your github notifications
2. in that last link you have this line https://github.com/lightbend/kafka-streams-query/blob/develop/examples/example-dsl/src/main/scala/com/lightbend/kafka/scala/iq/example/serializers/AppSerializers.scala#L25 am I right in saying you can do that because LogRecordAvro was created by the avro code generation tool and therefore meets the condition of T <: org.apache.avro.specific.SpecificRecord on the SpecificAvroSerializerWithSchemaRegistry ?

thats the bit I can't get my head around, I am not using code generation, just have a plain old Case Class that I want to create a SpecificAvroSerdeWithSchemaRegistry for. Maybe bijection is the answer from a brief look at their repo

from kafka-streams-scala.

I think we can discuss here.

Regarding (2) you are correct - I was using code generation using the sbt plugin for avro (which keeps it nicely decoupled from the code base). If you are not using code generation maybe bijection is the answer.

from kafka-streams-scala.

I am trending towards using code generation now. its a poc I am working on currently and would need to go back and reevaluate the ci process but it will probably end up being for the best. Thank-you for all the help.

from kafka-streams-scala.