title: Wrong serializers used on aggregate
body: <p>I am working of a kafka-streams application in which I process log events. In this case I want to aggregate WorkflowInput types into a Workflow type. I am having issues getting the aggregate working.</p>
  
  <pre><code>final KStream&lt;String, WorkflowInput&gt; filteredStream = someStream;
  final KTable&lt;String, Workflow&gt; aggregatedWorkflows = filteredStream
      .peek((k, v) -&gt; {
          if (!(v instanceof WorkflowInput)) {
              throw new AssertionError("Type not expected");
          }
      })
      .groupByKey()
      .&lt;Workflow&gt;aggregate(Workflow::new, (k, input, workflow) -&gt; workflow.updateFrom(input),
              Materialized.&lt;String, Workflow, KeyValueStore&lt;Bytes, byte[]&gt;&gt;as("worflow-cache")
                  .withKeySerde(Serdes.String())
                  .withValueSerde(Serdes.serdeFrom(new JsonSerializer&lt;Workflow&gt;(), new JsonDeserializer&lt;Workflow&gt;(Workflow.class)));
  </code></pre>
  
  <p>I get the following exception:Caused by: <code>org.apache.kafka.streams.errors.StreamsException: A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.apache.kafka.common.serialization.StringSerializer) is not compatible to the actual key or value type (key type: java.lang.String / value type: workflowauditstreamer.WorkflowInput).</code></p>
  
  <p>Two things to notice:
   * The value serializer is a StringSerializer, while I configured something different using <code>withValueSerde</code>.
   * The actual value type is <code>WorkflowInput</code> while I expect <code>Workflow</code> since that is my aggregated value type.</p>
  
  <p>I am new to kafka-streams so I might be missing something obvious, but I cannot figure it out. What am I missing here?</p>