Comments (9)
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.
- feel free to tell me this isnt the place for this convo, dont want to clutter up your github notifications
- 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 theSpecificAvroSerializerWithSchemaRegistry
?
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.
Related Issues (20)
- Issue with KStreams transform returning null HOT 1
- Is 2.11 supported? HOT 2
- KGroupedStreamS recursion HOT 12
- case classes HOT 12
- Implicit Serdes HOT 22
- StateStore API HOT 1
- Mockedstreams for unit testing! HOT 2
- KTable#filter(K,V => Boolean, Materialized) is called recusively on KTableS HOT 2
- Kafka server utils not being packaged HOT 7
- java.lang.ClassCastException for reduce api call HOT 11
- Timestamp extractor not supported HOT 1
- KStreamS#merge is recursive, causing StackOverflow on initialization HOT 3
- KStream-KTable inner join HOT 1
- Proposal to add Circe support HOT 4
- support for ValueTransformerWithKey HOT 3
- kafka streams doesn't rebalance when one broker is down HOT 1
- Kafka 2.0 support HOT 2
- stable URL to publish 'current' API documentation HOT 2
- Archive this repository HOT 3
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from kafka-streams-scala.