Thanks, Guozhang, I'll file a JIRA this afternoon. --Fred
On Wed, May 18, 2016 at 11:51 AM, Guozhang Wang <wangg...@gmail.com> wrote:
> Fred,
>
> You are right, only default serdes provided through configs are
> auto-configured today. But we could auto-configure other serdes passed
> along side the topology builder as well. Do you want to file a JIRA to keep
> track of this?
>
>
> Guozhang
>
> On Wed, May 18, 2016 at 11:36 AM, Fred Patton <thoughtp...@gmail.com>
> wrote:
>
> > Thanks, so much, Guozhang. It didn't work immediately when I tried end of
> > day, so I deferred til this morning. It seems I also needed to invoke
> > configure manually, or pass the props to the deserializer constructor. I
> am
> > providing the revised example in case it helps any one. Thank you Liquan
> > for pointing out the need for the cast, which was also the case.
> >
> > Regards,
> > Fred
> >
> > def main(args: Array[String]) {
> >
> > val schemaRegistry = new CachedSchemaRegistryClient("
> > http://localhost:8081";,
> > AbstractKafkaAvroSerDeConfig.MAX_SCHEMAS_PER_SUBJECT_DEFAULT)
> >
> > val builder: KStreamBuilder = new KStreamBuilder
> >
> > val streamingConfig = {
> >
> > val settings = new Properties
> >
> > settings.put(StreamsConfig.JOB_ID_CONFIG, "kstreams")
> >
> > settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
> > "localhost:9092")
> >
> > settings.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG,
> > "localhost:2181")
> >
> >
> settings.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
> > "http://localhost:8081";);
> >
> > settings.put(StreamsConfig.KEY_SERIALIZER_CLASS_CONFIG,
> > "io.confluent.kafka.serializers.KafkaAvroSerializer")
> >
> > settings.put(StreamsConfig.KEY_DESERIALIZER_CLASS_CONFIG,
> > "io.confluent.kafka.serializers.KafkaAvroDeserializer")
> >
> > settings.put(StreamsConfig.VALUE_SERIALIZER_CLASS_CONFIG,
> > "io.confluent.kafka.serializers.KafkaAvroSerializer")
> >
> > settings.put(StreamsConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
> > "io.confluent.kafka.serializers.KafkaAvroDeserializer")
> >
> >
> settings.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG,
> > "true")
> >
> > settings
> >
> > }
> >
> >
> > import KeyValueImplicits._
> >
> >
> > val stringSerializer = new StringSerializer
> >
> > val stringDeserializer = new StringDeserializer
> >
> > val bcAccessLogSerializer = new
> > SpecificAvroSerializer[BcAccessLog](schemaRegistry)
> >
> > val bcAccessLogDeserializer = new
> > SpecificAvroDeserializer[BcAccessLog](schemaRegistry)
> >
> >
> >
> >
> >
> bcAccessLogSerializer.configure(streamingConfig.asInstanceOf[java.util.Map[String,
> > String]], false)
> >
> >
> >
> >
> bcAccessLogDeserializer.configure(streamingConfig.asInstanceOf[java.util.Map[String,
> > String]], false)
> >
> >
> > val rawSourceStream: KStream[String, String] =
> > builder.stream(stringDeserializer,
> >
> > stringDeserializer,
> >
> > "raw-accesslog-" + testRun)
> >
> >
> > val filteredStream: KStream[String, BcAccessLog] =
> > rawSourceStream.mapValues(parseAsJson)
> >
> > .filter(isCdcEvent)
> >
> > .map((key, value) => (getAccessLogKey(value),
> toBcAccessLog(value)))
> >
> >
> > filteredStream
> >
> > .through("accesslog-" + testRun,
> >
> > stringSerializer,
> >
> > bcAccessLogSerializer,
> >
> > stringDeserializer,
> >
> > bcAccessLogDeserializer)
> >
> > .mapValues(value => {
> >
> > println("filteredSourceStream value: " + value)
> >
> > value
> >
> > })
> >
> > .process(new CdcProcessorSupplier)
> >
> >
> > val stream: KafkaStreams = new KafkaStreams(builder, streamingConfig)
> >
> > println("\nstart filtering BcAccessLog test run: " + testRun + "\n")
> >
> > stream.start()
> >
> > }
> >
> > On Tue, May 17, 2016 at 5:29 PM, Guozhang Wang <wangg...@gmail.com>
> wrote:
> >
> > > Hello Fred,
> > >
> > > Thanks for reporting this. I looked through your code and found this is
> > due
> > > to a bug in
> io.confluent.examples.streams.utils.SpecificAvroDeserializer.
> > >
> > > I will fix this bug in Confluent's repo, as for you to work around it,
> > you
> > > can create your own SpecificAvroDeserializer with the correct logic by
> > > copy-past Confluent class with the following one-line change:
> > >
> > > In configure(Map<String, ?> configs, boolean isKey) function, add the
> > > following key-value pair to the configs before calling inner.configure:
> > >
> > > configs.put("specific.avro.reader", "true");
> > >
> > >
> > > Guozhang
> > >
> > > On Tue, May 17, 2016 at 11:35 AM, Fred Patton <thoughtp...@gmail.com>
> > > wrote:
> > >
> > > > Before, switching the example from to() to through() things seemed to
> > > work
> > > > fine. Afterward, I get ClassCastExceptions from GenericRecord to
> > > > SpecificRecord. Hopefully, someone can point out something quick and
> > dumb
> > > > so I can get my demo wrapped up. I never get past the through()
> > method. I
> > > > am using the Kafka Streams tech preview, and using Avro compiled from
> > > > AvroTools. Additionally, initially I had a second parallel
> application
> > > > taking the output of this to() invocation and continuing with the
> same
> > > > logic, there was no crash, but no code ever ran such as initializing
> > the
> > > > context. Any illumination is most appreciated.
> > > >
> > > > Here's the stack trace followed by the offending bit code...
> > > >
> > > > [info] CdcProcessorSupplier::get
> > > > [info] CdcProcessor::init - ctx:
> > > >
> > org.apache.kafka.streams.processor.internals.ProcessorContextImpl@c5ff552
> > > > [info] CdcProcessor::close
> > > > [error] Exception in thread "StreamThread-1"
> > > java.lang.ClassCastException:
> > > > org.apache.avro.generic.GenericData$Record cannot be cast to
> > > > org.apache.avro.specific.SpecificRecord
> > > > [error] at
> > > >
> > > >
> > >
> >
> io.confluent.examples.streams.utils.SpecificAvroDeserializer.deserialize(SpecificAvroDeserializer.java:51)
> > > > [error] at
> > > >
> > > >
> > >
> >
> io.confluent.examples.streams.utils.SpecificAvroDeserializer.deserialize(SpecificAvroDeserializer.java:24)
> > > > [error] at
> > > >
> > > >
> > >
> >
> org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:41)
> > > > [error] at
> > > >
> > > >
> > >
> >
> org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:77)
> > > > [error] at
> > > >
> > > >
> > >
> >
> org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:113)
> > > > [error] at
> > > >
> > > >
> > >
> >
> org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:134)
> > > > [error] at
> > > >
> > > >
> > >
> >
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:334)
> > > > [error] at
> > > >
> > > >
> > >
> >
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:248)
> > > >
> > > > def main(args: Array[String]) {
> > > > val schemaRegistry = new CachedSchemaRegistryClient("
> > > > http://localhost:8081";,
> > > > AbstractKafkaAvroSerDeConfig.MAX_SCHEMAS_PER_SUBJECT_DEFAULT)
> > > > val builder: KStreamBuilder = new KStreamBuilder
> > > > val streamingConfig = {
> > > > val settings = new Properties
> > > > settings.put(StreamsConfig.JOB_ID_CONFIG, "kstreams")
> > > > settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
> > > > "localhost:9092")
> > > > settings.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG,
> > > > "localhost:2181")
> > > > settings.put(StreamsConfig.KEY_SERIALIZER_CLASS_CONFIG,
> > > > "io.confluent.kafka.serializers.KafkaAvroSerializer")
> > > > settings.put(StreamsConfig.KEY_DESERIALIZER_CLASS_CONFIG,
> > > > "io.confluent.kafka.serializers.KafkaAvroDeserializer")
> > > > settings.put(StreamsConfig.VALUE_SERIALIZER_CLASS_CONFIG,
> > > > "io.confluent.kafka.serializers.KafkaAvroSerializer")
> > > > settings.put(StreamsConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
> > > > "io.confluent.kafka.serializers.KafkaAvroDeserializer")
> > > > settings
> > > > }
> > > >
> > > > import KeyValueImplicits._
> > > >
> > > > val stringSerializer = new StringSerializer
> > > > val stringDeserializer = new StringDeserializer
> > > > val bcAccessLogSerializer = new
> > > > SpecificAvroSerializer[BcAccessLog](schemaRegistry)
> > > > val bcAccessLogDeserializer = new
> > > > SpecificAvroDeserializer[BcAccessLog](schemaRegistry)
> > > >
> > > > val rawSourceStream: KStream[String, String] =
> > > > builder.stream(stringDeserializer,
> > > > stringDeserializer,
> > > > "raw-accesslog-" + testRun)
> > > >
> > > > val filteredStream: KStream[String, BcAccessLog] =
> > > > rawSourceStream.mapValues(parseAsJson)
> > > > .filter(isCdcEvent)
> > > > .map((key, value) => (getAccessLogKey(value),
> > > toBcAccessLog(value)))
> > > >
> > > > filteredStream
> > > > .through("accesslog-" + testRun,
> > > > stringSerializer,
> > > > bcAccessLogSerializer,
> > > > stringDeserializer,
> > > > bcAccessLogDeserializer)
> > > > .mapValues(value => {
> > > > println("filteredSourceStream value: " + value)
> > > > value
> > > > })
> > > > .process(new CdcProcessorSupplier)
> > > >
> > > > val stream: KafkaStreams = new KafkaStreams(builder,
> > streamingConfig)
> > > > println("\nstart filtering BcAccessLog test run: " + testRun +
> > "\n")
> > > > stream.start()
> > > > }
> > > >
> > > > Regards,
> > > > Fred Patton
> > > >
> > >
> > >
> > >
> > > --
> > > -- Guozhang
> > >
> >
>
>
>
> --
> -- Guozhang
>
