title: Can Kafka Streams consume message in a format and produce another format such as AVRO message
body: <p>I am using kafka streams to consume JSON string from one topic, process and generate response to be stored in another topic. However the message that needs to be produced to the response topic needs to be in avro format.</p>
  
  <p>I have tried using key as string serde and value as <strong>SpecificAvroSerde</strong></p>
  
  <p>Following is my Code to create Topology:</p>
  
  <pre><code>StreamsBuilder builder = new StreamsBuilder();
  KStream&lt;Object, Object&gt; consumerStream =builder.stream(kafkaConfiguration.getConsumerTopic());
  consumerStream = consumerStream.map(getKeyValueMapper(keyValueMapperClassName));
  consumerStream.to(kafkaConfiguration.getProducerTopic());
  </code></pre>
  
  <p>Following is my config</p>
  
  <pre><code>    if (schemaRegistry != null &amp;&amp; schemaRegistry.length &gt; 0) {
          streamsConfig.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, String.join(",", schemaRegistry));          
      }
      streamsConfig.put(this.keySerializerKeyName, StringSerde.class);
      streamsConfig.put(this.valueSerialzerKeyName, SpecificAvroSerde.class);
      streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
      streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
      streamsConfig.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, batchSize);
      streamsConfig.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, FailOnInvalidTimestamp.class);
      streamsConfig.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, processingGuarantee);
      streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, Integer.parseInt(commitIntervalMs));
      streamsConfig.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numberOfThreads);
      streamsConfig.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, replicationFactor);
      streamsConfig.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
      streamsConfig.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, ProductionExceptionHandler.class);
      streamsConfig.put(StreamsConfig.TOPOLOGY_OPTIMIZATION,StreamsConfig.OPTIMIZE);
      streamsConfig.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionMode);
      streamsConfig.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
  </code></pre>
  
  <p>I am seeing the following error when I try with the example:</p>
  
  <pre><code>org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
  Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!
  </code></pre>