title: How to join two Kafka streams and produce the result in a topic with Avro values
body: <p>I've got two Kafka Streams with keys in <code>String</code> and values in <code>Avro</code> format which I have created using KSQL. </p>
  
  <p>Here's the first one:</p>
  
  <pre><code>DESCRIBE EXTENDED STREAM_1; 
  Type                 : STREAM
  Key field            : IDUSER
  Timestamp field      : Not set - using &lt;ROWTIME&gt;
  Key format           : STRING
  Value format         : AVRO
  Kafka output topic   : STREAM_1 (partitions: 4, replication: 1)
  
   Field                      | Type
  --------------------------------------------------------
   ROWTIME                    | BIGINT           (system)
   ROWKEY                     | VARCHAR(STRING)  (system)
   FIRSTNAME                  | VARCHAR(STRING)
   LASTNAME                   | VARCHAR(STRING)
   IDUSER                     | VARCHAR(STRING)
  </code></pre>
  
  <p>and the second one: </p>
  
  <pre><code>DESCRIBE EXTENDED STREAM_2;
  Type                 : STREAM
  Key field            : IDUSER
  Timestamp field      : Not set - using &lt;ROWTIME&gt;
  Key format           : STRING
  Value format         : AVRO
  Kafka output topic   : STREAM_2 (partitions: 4, replication: 1)
  
   Field                      | Type
  --------------------------------------------------------
   ROWTIME                    | BIGINT           (system)
   ROWKEY                     | VARCHAR(STRING)  (system)
   USERNAME                   | VARCHAR(STRING)
   IDUSER                     | VARCHAR(STRING)
   DEVICE                     | VARCHAR(STRING)
  </code></pre>
  
  <p>The desired output should include <code>IDUSER</code>, <code>LASTNAME</code>, <code>DEVICE</code> and <code>USERNAME</code>.</p>
  
  <p>I want to <code>left join</code> these streams (on <code>IDUSER</code>) using Streams API and write the output into a kafka topic. </p>
  
  <p>To do so, I've tried the following: </p>
  
  <pre><code>public static void main(String[] args) {
  
      final Properties streamsConfiguration = new Properties();
  
      streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-strteams");
      streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
      streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "localhost:2181");
      streamsConfiguration.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
  
      streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
      streamsConfiguration.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
      streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  
      final Serde&lt;String&gt; stringSerde = Serdes.String();
      final Serde&lt;GenericRecord&gt; genericAvroSerde = new GenericAvroSerde();
  
  
      boolean isKeySerde = false;
      genericAvroSerde.configure(Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081"), isKeySerde);
  
      KStreamBuilder builder = new KStreamBuilder();
  
      KStream&lt;String, GenericRecord&gt; left = builder.stream("STREAM_1");
      KStream&lt;String, GenericRecord&gt; right = builder.stram("STREAM_2");
  
      // Java 8+ example, using lambda expressions
      KStream&lt;String, GenericRecord&gt; joined = left.leftJoin(right,
          (leftValue, rightValue) -&gt; "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
          JoinWindows.of(TimeUnit.MINUTES.toMillis(5)),
          Joined.with(
            stringSerde, /* key */
            genericAvroSerde,   /* left value */
            genericAvroSerde)  /* right value */
        );
      joined.to(stringSerde, genericAvroSerde, "streams-output-testing");
  
      KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration);
      streams.cleanUp();
      streams.start();
  
      Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
  }
  </code></pre>
  
  <p>However, </p>
  
  <pre><code>KStream&lt;String, GenericRecord&gt; joined = ...
  </code></pre>
  
  <p>throws an error on my IDE: </p>
  
  <pre><code>incompatible types: inference variable VR has incompatible bounds
  </code></pre>
  
  <p>When I try to use a <code>String</code> <code>Serde</code> for both keys and values, it works but the data is not that readable from <code>kafka-console-consumer</code>. What I want to do is to produce the data in AVRO format in order to be able to read them off using <code>kafka-avro-console-consumer</code>. </p>