title: Creating a state store with Avro inside Kstream Consumer Processor
body: <p>I have a consumer defined as below. It reads a avro message out of topic and constructs a statestore of aggregated data, which is also of type avro.</p>
  <pre><code>    @Bean
      public Consumer&lt;KStream&lt;String, InputEvent&gt;&gt; avroTest() {
          Serde&lt;OutputEvent&gt; serdeOutEvent = new SpecificAvroSerde&lt;&gt;(schemaRegistryClient);
          return st -&gt; st.groupByKey().aggregate(OutputEvent::new, (key, currentEvent, outputEvent) -&gt; {
              //aggregate here
              return outputEvent;
          }, Materialized.with(new Serdes.StringSerde(), serdeOutEvent).toStream();
      }
  </code></pre>
  <p>The function is able to read messages from topic and create the first aggregated result, but when it tries to store it in statestore, receives a 404 for schema not present.</p>
  <pre><code>Exception in thread &quot;odoAvroTest-e4ef8e3e-ea1e-458c-b309-b2afefbeacec-StreamThread-1&quot; org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000, topic=odometer, partition=0, offset=0, stacktrace=org.apache.kafka.common.errors.SerializationException: Error retrieving Avro schema: {&quot;type&quot;:&quot;record&quot;,&quot;name&quot;: &quot;&quot; .... }
  Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Subject not found.; error code: 40401
      at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:226)
      at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:252)
      at io.confluent.kafka.schemaregistry.client.rest.RestService.lookUpSubjectVersion(RestService.java:319)
      at io.confluent.kafka.schemaregistry.client.rest.RestService.lookUpSubjectVersion(RestService.java:307)
      at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getIdFromRegistry(CachedSchemaRegistryClient.java:165)
      at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getId(CachedSchemaRegistryClient.java:297)
      at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:73)
      at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:53)
      at io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer.serialize(SpecificAvroSerializer.java:65)
      at io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer.serialize(SpecificAvroSerializer.java:38)
      at org.apache.kafka.streams.state.internals.ValueAndTimestampSerializer.serialize(ValueAndTimestampSerializer.java:59)
      at org.apache.kafka.streams.state.internals.ValueAndTimestampSerializer.serialize(ValueAndTimestampSerializer.java:50)
      at org.apache.kafka.streams.state.internals.ValueAndTimestampSerializer.serialize(ValueAndTimestampSerializer.java:27)
      at org.apache.kafka.streams.state.StateSerdes.rawValue(StateSerdes.java:192)
      at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.put(MeteredKeyValueStore.java:166)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl$KeyValueStoreReadWriteDecorator.put(ProcessorContextImpl.java:486)
      at org.apache.kafka.streams.kstream.internals.KStreamAggregate$KStreamAggregateProcessor.process(KStreamAggregate.java:103)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87)
      at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:363)
      at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:199)
      at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:425)
      at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:912)
      at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:819)
      at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:788)
  </code></pre>
  <p>Do let know if there are additional config tweaks that are necessary to make this work. When i change the input to hashmap and /or a simple POJO amd use JSONSerde, the code seems to work and creates aggregation</p>