title: Spring Cloud Kafka: Can't serialize data for output stream when two processors are active
body: <p>I have a working setup for Spring Cloud Kafka Streams with functional programming style.
  There are two use cases, which are configured via <code>application.properties</code>.
  Both of them work individually, but as soon as I activate both at the same time, I get a serialization error for the output stream of the second use case:</p>
  <pre class="lang-none prettyprint-override"><code>Exception in thread &quot;ActivitiesAppId-05296224-5ea1-412a-aee4-1165870b5c75-StreamThread-1&quot; org.apache.kafka.streams.errors.StreamsException:
  Error encountered sending record to topic outputActivities for task 0_0 due to:
  ...
  Caused by: org.apache.kafka.common.errors.SerializationException:
  Can't serialize data [com.example.connector.model.Activity@497b37ff] for topic [outputActivities]
  Caused by: com.fasterxml.jackson.databind.exc.InvalidDefinitionException:
  Incompatible types: declared root type ([simple type, class com.example.connector.model.Material]) vs com.example.connector.model.Activity
  </code></pre>
  <p>The last line here is important, as the &quot;declared root type&quot; is from the <code>Material</code> class, but not the <code>Activity</code> class, which is probably the source error.</p>
  <p>Again, when I only activate the second use case before starting the application, everything works fine. So I assume that the &quot;Material&quot; processor somehow interfers with the &quot;Activities&quot; processor (or its serializer), but I don't know when and where.</p>
  <hr />
  <p><strong>Setup</strong></p>
  <p>1.) use case: &quot;Materials&quot;</p>
  <ul>
  <li>one input stream -&gt; transformation -&gt; one output stream</li>
  </ul>
  <pre class="lang-java prettyprint-override"><code>@Bean
  public Function&lt;KStream&lt;String, MaterialRaw&gt;, KStream&lt;String, Material&gt;&gt; processMaterials() {...}
  </code></pre>
  <p><code>application.properties</code></p>
  <pre class="lang-properties prettyprint-override"><code>spring.cloud.stream.kafka.streams.binder.functions.processMaterials.applicationId=MaterialsAppId
  spring.cloud.stream.bindings.processMaterials-in-0.destination=inputMaterialsRaw
  spring.cloud.stream.bindings.processMaterials-out-0.destination=outputMaterials
  </code></pre>
  <p>2.) use case: &quot;Activities&quot;</p>
  <ul>
  <li>two input streams -&gt; joining -&gt; one output stream</li>
  </ul>
  <pre class="lang-java prettyprint-override"><code>@Bean
  public BiFunction&lt;KStream&lt;String, ActivityRaw&gt;, KStream&lt;String, Assignee&gt;, KStream&lt;String, Activity&gt;&gt; processActivities() {...}
  </code></pre>
  <p><code>application.properties</code></p>
  <pre class="lang-properties prettyprint-override"><code>spring.cloud.stream.kafka.streams.binder.functions.processActivities.applicationId=ActivitiesAppId
  spring.cloud.stream.bindings.processActivities-in-0.destination=inputActivitiesRaw
  spring.cloud.stream.bindings.processActivities-in-1.destination=inputAssignees
  spring.cloud.stream.bindings.processActivities-out-0.destination=outputActivities
  </code></pre>
  <p>The two processors are also defined as stream function in <code>application.properties</code>: <code>spring.cloud.stream.function.definition=processActivities;processMaterials</code></p>
  <p>Thanks!</p>
  <p>Update - Here's how I use the processors in the code:</p>
  <p><strong>Implementation</strong></p>
  <pre class="lang-java prettyprint-override"><code>// Material model
  @Getter
  @Setter
  @AllArgsConstructor
  @NoArgsConstructor
  public class MaterialRaw {
      private String id;
      private String name;
  }
  
  @Getter
  @Setter
  @AllArgsConstructor
  @NoArgsConstructor
  public class Material {
      private String id;
      private String name;
  }
  
  // Material processor
  @Bean
  public Function&lt;KStream&lt;String, MaterialRaw&gt;, KStream&lt;String, Material&gt;&gt; processMaterials() {
      return materialsRawStream -&gt; materialsRawStream .map((recordKey, materialRaw) -&gt; {
          // some transformation
          final var newId = materialRaw.getId() + &quot;---foo&quot;;
          final var newName = materialRaw.getName() + &quot;---bar&quot;;
          final var material = new Material(newId, newName);
  
          // output
          return new KeyValue&lt;&gt;(recordKey, material); 
      };
  }
  </code></pre>
  <pre class="lang-java prettyprint-override"><code>// Activity model
  @Getter
  @Setter
  @AllArgsConstructor
  @NoArgsConstructor
  public class ActivityRaw {
      private String id;
      private String name;
  }
  
  @Getter
  @Setter
  @AllArgsConstructor
  @NoArgsConstructor
  public class Assignee {
      private String id;
      private String assignedAt;
  }
  
  /**
   * Combination of `ActivityRaw` and `Assignee`
   */
  @Getter
  @Setter
  @AllArgsConstructor
  @NoArgsConstructor
  public class Activity {
      private String id;
      private Integer number;
      private String assignedAt;
  }
  
  // Activity processor
  @Bean
  public BiFunction&lt;KStream&lt;String, ActivityRaw&gt;, KStream&lt;String, Assignee&gt;, KStream&lt;String, Activity&gt;&gt; processActivities() {
      return (activitiesRawStream, assigneesStream) -&gt; { 
          final var joinWindow = JoinWindows.of(Duration.ofDays(30));
  
          final var streamJoined = StreamJoined.with(
              Serdes.String(),
              new JsonSerde&lt;&gt;(ActivityRaw.class),
              new JsonSerde&lt;&gt;(Assignee.class)
          );
  
          final var joinedStream = activitiesRawStream.leftJoin(
              assigneesStream,
              new ActivityJoiner(),
              joinWindow,
              streamJoined
          );
  
          final var mappedStream = joinedStream.map((recordKey, activity) -&gt; {
              return new KeyValue&lt;&gt;(recordKey, activity);
          });
  
          return mappedStream;
      };
  }
  </code></pre>