title: Kafka streams - joining two ktables invokes join function twice
body: <p>i am trying to join 2 KTables.</p>
  
  <pre><code>KTable&lt;String, RecordBean&gt; recordsTable = builder.table(Serdes.String(),
      new JsonPOJOSerde&lt;&gt;(RecordBean.class),
      bidTopic, RECORDS_STORE);
  
  KTable&lt;String, ImpressionBean&gt; impressionsTable = builder.table(Serdes.String(),
      new JsonPOJOSerde&lt;&gt;(ImpressionBean.class),
      impressionTopic, IMPRESSIONS_STORE);
  
  KTable&lt;String, RecordBean&gt; mergedByTxId = recordsTable
      .join(impressionsTable, merge());
  </code></pre>
  
  <p>The merge functions is very simple, i am just copying value from one bean to another.</p>
  
  <pre><code>public static &lt;K extends BidInfo, V extends BidInfo&gt; ValueJoiner&lt;K, V, K&gt; merge() {
  return (v1, v2) -&gt; {
    v1.setRtbWinningBidAmount(v2.getRtbWinningBidAmount());
    return v1;
  };
  </code></pre>
  
  <p>But for some reasons the join function is calling twice on single produced record.
  Please see streaming/producer config below</p>
  
  <pre><code>Properties streamsConfiguration = new Properties();
  streamsConfiguration
      .put(StreamsConfig.APPLICATION_ID_CONFIG, "join-impressions");
  streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
  
  streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, CLUSTER.zookeeperConnect());
  streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, folder.newFolder("kafka-streams-tmp")
      .getAbsolutePath());
  
  return streamsConfiguration;
  </code></pre>
  
  <p>Producer config - </p>
  
  <pre><code>Properties producerConfig = new Properties();
  producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
  producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
  producerConfig.put(ProducerConfig.RETRIES_CONFIG, 0);
  producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  
  return producerConfig;
  </code></pre>
  
  <p>Next i am submitting single record per stream. Both records has same keys.
  I am expecting to receive single record as output.</p>
  
  <pre><code> IntegrationTestUtils.produceKeyValuesSynchronously(bidsTopic,
      Arrays.asList(new KeyValue("1", getRecordBean("1"))),
      getProducerProperties());
  
  IntegrationTestUtils.produceKeyValuesSynchronously(impressionTopic,
      Arrays.asList(new KeyValue("1", getImpressionBean("1"))),
      getProducerProperties());
  
  List&lt;KeyValue&lt;String, String&gt;&gt; parsedRecord =
      IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(getConsumerProperties(),
          outputTopic, 1);
  </code></pre>
  
  <p>But ValueJoiner triggers 2 times, and i am getting 2 identical output records instead one. During trigger time - both values from both streams exist - and i cannot get what is triggering second execution.</p>
  
  <p>Without joining - i cannot reproduce this behavior. 
  I cannot find any working example of 2 ktable join - so cannot understand whats wrong with my approach.</p>
  
  <p>Adding simple code that demonstrate same behavior</p>
  
  <pre><code>KStreamBuilder builder = new KStreamBuilder();
  
  KTable&lt;String, String&gt; first = builder.table("stream1", "storage1");
  KTable&lt;String, String&gt; second = builder.table("stream2", "storage2");
  
  KTable&lt;String, String&gt; joined = first.join(second, (value1, value2) -&gt; value1);
  
  joined.to("output");
  
  KafkaStreams streams = new KafkaStreams(builder, getStreamingProperties());
  
  streams.start();
  
  IntegrationTestUtils.produceKeyValuesSynchronously("stream1",
      Arrays.asList(new KeyValue("1", "first stream")),
      getProducerProperties());
  
  IntegrationTestUtils.produceKeyValuesSynchronously("stream2",
      Arrays.asList(new KeyValue("1", "second stream")),
      getProducerProperties());
  
  List&lt;KeyValue&lt;String, String&gt;&gt; parsedRecord =
      IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(getConsumerProperties(),
          "output", 1);
  </code></pre>