title: Complex Object KStream GlobalKTable left-join
body: <p>I am new to Kafka Streams. I want to perform following KStream-GlobalKTable pure DSL based left-join operation, and not using map operation.</p>
  <p>I have an input stream <code>a.topic</code> which is &lt;String, A&gt;, where value :</p>
  <pre><code>{
      &quot;b_obj&quot;: {
          &quot;b_value&quot;: &quot;xyz&quot;,
          &quot;c_list&quot;: [
              {
                  &quot;d_obj&quot;: {
                      &quot;d_id1&quot;: &quot;value1&quot;,
                      &quot;d_id2&quot;: &quot;value2&quot;,
                      &quot;d_value&quot;: &quot;some value&quot;
                  },
                  &quot;c_value&quot;: &quot;jkl&quot;
              },
              {
                  &quot;d_obj&quot;: {
                      &quot;d_id1&quot;: &quot;value3&quot;,
                      &quot;d_id2&quot;: &quot;value4&quot;,
                      &quot;d_value&quot;: &quot;some value 2&quot;
                  },
                  &quot;c_value&quot;: &quot;pqr&quot;
              }
          ]
      },
      &quot;a_value&quot;: &quot;abcd&quot;
  }
  </code></pre>
  <p>And another input topic <code>e.topic</code> which is &lt;String, E&gt;, where value :</p>
  <pre><code>{
      &quot;e_id1&quot;: &quot;value1&quot;,
      &quot;e_id2&quot;: &quot;value2&quot;,
      &quot;e_value&quot;: &quot;some value&quot;
  }
  </code></pre>
  <p>I want to perform left join operation <code>a.topic</code> is a stream, and master data <code>e.topic</code> is a global table to achieve result value as</p>
  <pre><code>{
      &quot;b_obj&quot;: {
          &quot;b_value&quot;: &quot;xyz&quot;,
          &quot;c_list&quot;: [
              {
                  &quot;d_obj&quot;: {
                      &quot;d_id1&quot;: &quot;value1&quot;,
                      &quot;d_id2&quot;: &quot;value2&quot;,
                      &quot;d_value&quot;: &quot;some value&quot;
                  },
                  &quot;e_obj&quot;: {
                      &quot;e_id1&quot;: &quot;value1&quot;,
                      &quot;e_id2&quot;: &quot;value2&quot;,
                      &quot;e_value&quot;: &quot;some value a&quot;
                  },
                  &quot;c_value&quot;: &quot;jkl&quot;
              },
              {
                  &quot;d_obj&quot;: {
                      &quot;d_id1&quot;: &quot;value3&quot;,
                      &quot;d_id2&quot;: &quot;value4&quot;,
                      &quot;d_value&quot;: &quot;some value 2&quot;
                  },
                  &quot;e_obj&quot;: {
                      &quot;e_id1&quot;: &quot;value3&quot;,
                      &quot;e_id2&quot;: &quot;value4&quot;,
                      &quot;e_value&quot;: &quot;some value b&quot;
                  },
                  &quot;c_value&quot;: &quot;pqr&quot;
              }
          ]
      },
      &quot;a_value&quot;: &quot;abcd&quot;
  }
  </code></pre>
  <p>and the join <strong>condition</strong> is <code>a.b.c[i].d.d_id1 == e.e_id1 &amp;&amp; a.b.c[i].d.d_id2 == e.e_id2</code></p>
  <p><strong>CODE</strong>:</p>
  <pre><code>public class ComplexBeanStream {
  
      public static void main(String[] args) {
  
          Serde&lt;A&gt; aSerde = new JsonSerde&lt;&gt;(A.class);
  
          Serde&lt;E&gt; eSerde = new JsonSerde&lt;&gt;(E.class);
  
          final Properties streamsConfiguration = new Properties();
          streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, &quot;complex-bean-app&quot;);
          streamsConfiguration.put(StreamsConfig.CLIENT_ID_CONFIG, &quot;complex-bean-client&quot;);
          streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;localhost:9092&quot;);
  
          StreamsBuilder builder = new StreamsBuilder();
  
          final GlobalKTable&lt;String, E&gt; eGlobalTable =
                  builder.globalTable(
                          &quot;e.topic&quot;,
                          Materialized.&lt;String, E, KeyValueStore&lt;Bytes, byte[]&gt;&gt;
                                  as(&quot;E-STORE&quot;)
                                  .withKeySerde(Serdes.String())
                                  .withValueSerde(eSerde)
                  );
  
          final KStream&lt;String, A&gt; aStream =
                  builder.stream(
                          &quot;a.topic&quot;,
                          Consumed.with(Serdes.String(), aSerde));
  
          // perform left-join here
  
          Topology topology = builder.build();
  
          System.out.println(&quot;\n\nComplexBeanStream Topology: \n&quot; + topology.describe());
  
          final KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration);
  
          streams.cleanUp();
  
          streams.start();
  
          Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
      }
  
  }
  
  class A {
  
      private B b_obj;
  
      private String a_value;
  }
  
  class B {
  
      private List&lt;C&gt; c_list;
  
      private String b_value;
  }
  
  class C {
  
      private D d_obj;
  
      private E e_obj;
  
      private String c_value;
  }
  
  class D {
  
      private String d_id1;
  
      private String d_id2;
  
      private String d_value;
  }
  
  class E {
  
      private String e_id1;
  
      private String e_id2;
  
      private String e_value;
  }
  </code></pre>