title: Kafka Stream-GlobalKTable join on a specific field
body: <p>So I have a KStream that that gets deserialized into a POJO like so</p>
  <pre><code>public class FinancialMessage {
  
  public String user_id;
  public String stock_symbol;
  public String exchange_id;
  
  }
  </code></pre>
  <p>And here's how the Global Ktable record looks like</p>
  <pre><code>public class CompanySectors {
  
  public String company_id;
  public String company_name;
  public String tckr;
  public String sector_cd;
  }
  </code></pre>
  <p>I want to be able to join the KStream's <code>stock_symbol</code> field with the Ktable's <code>tckr</code> field. Is this possible? I want to create a new <code>EnrichedMessage</code> object before I stream it into another topic. I had code like below but I seem to be getting some null pointer exceptions.</p>
  <pre><code>Exception in thread &quot;trade-enrichment-stream-0c7e7782-4217-4450-8086-21871b4ebc45-StreamThread-1&quot; java.lang.NullPointerException
      at com.domain.EnrichedMessage.&lt;init&gt;(EnrichedMessage.java:51)
      at com.domain.TradeEnrichmentTopology.lambda$3(TradeEnrichmentTopology.java:73)
      at org.apache.kafka.streams.kstream.internals.KStreamKTableJoinProcessor.process(KStreamKTableJoinProcessor.java:79)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$process$2(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      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:101)
      at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$3(StreamTask.java:383)
      at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:801)
      at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:383)
      at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:475)
      at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:550)
      at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:802)
      at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697)
      at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670)
  </code></pre>
  <p>Here's what the code snippet looks like.</p>
  <pre><code>KStream&lt;String, FinancialMessage&gt; financialMessageStream =
          builder.stream(
              INCOMING_TOPIC,
              Consumed.with(Serdes.String(), financialMessageSerde)
          );
  
      GlobalKTable&lt;String, CompanySectors&gt; companySectorsStore = 
          builder.globalTable(
              KTABLE_TOPIC,
              Consumed.with(Serdes.String(), companySectorsSerde)
      );
      
      KStream&lt;String, EnrichedMessage&gt; enrichedStream = financialMessageStream.leftJoin(
          companySectorsStore,
          (financialMessageKey, financialMessageValue) -&gt; financialMessageValue.stock_symbol,
          (financialMessageValue, companySectorsValue) -&gt; new EnrichedMessage(financialMessageValue, companySectorsValue)
      );
      
      enrichedStream.to(
          OUTGOING_TOPIC,
          Produced.with(Serdes.String(), enrichedMessageSerde));
  </code></pre>
  <p>I imagine that there might be some error in my leftJoin logic.</p>