title: KStream: Error Reading and Writing Avro records
body: <p>I’m trying to write avro record that I read from a topic into another topic, intentions it to augment it with transformation after I get this routing working. I have used the KStream with avro code from one of the example with some modifications to connect to Schema Registry for retrieving the avro schema. </p>
  
  <pre><code>streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "mysql-stream-processing");
  streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
  streamsConfiguration.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
  final Serde&lt;GenericRecord&gt; keySerde = new GenericAvroSerde(
          new CachedSchemaRegistryClient(schemaRegistryUrl, 100),
          Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
                  schemaRegistryUrl));
  final Serde&lt;GenericRecord&gt; valueSerde = new GenericAvroSerde(
          new CachedSchemaRegistryClient(schemaRegistryUrl, 100),
          Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
                  schemaRegistryUrl));
  streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);
  
  final KStreamBuilder builder = new KStreamBuilder();
  
  final KStream&lt;GenericRecord, GenericRecord&gt; record = builder.stream("dbserver1.employees.employees");
  
  record.print(keySerde, valueSerde);
  
  record.to(keySerde, valueSerde, "newtopic");
  
  
  record.foreach((key, val) -&gt; System.out.println(key.toString()+"  "+val.toString()));
  final KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration);
  
  streams.cleanUp();
  streams.start();
  </code></pre>
  
  <p>When run print() works as I can see the record in the console, but Im unable to get the record written to the “newtopic”, failing with the below error</p>
  
  <pre><code>Exception in thread "StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000, topic=dbserver1.employees.employees, partition=0, offset=0
          at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:217)
          at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:627)
          at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361)
  Caused by: org.apache.kafka.streams.errors.StreamsException: A serializer (key: io.confluent.examples.streams.utils.GenericAvroSerializer / value: io.confluent.examples.streams.utils.GenericAvroSerializer) is not compatible to the actual key or value type (key type: [B / value type: [B). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.
          at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:81)
          at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
          at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:70)
          at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:198)
          ... 2 more
  Caused by: java.lang.ClassCastException: [B cannot be cast to org.apache.avro.generic.GenericRecord
          at io.confluent.examples.streams.utils.GenericAvroSerializer.serialize(GenericAvroSerializer.java:25)
          at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:77)
          at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:79)
          ... 5 more
  </code></pre>