title: End-of-window outer join with KafkaStreams
body: <p>I have a Kafka topic where I expect messages with two different key types: old and new.
  i.e. <code>"1-new"</code>, <code>"1-old"</code>, <code>"2-new"</code>, <code>"2-old"</code>. Keys are unique, but some might be missing.</p>
  
  <p>Now using Kotlin and KafkaStreams API I can log those messages with have same key id from new and old.</p>
  
  <pre><code>    val windows = JoinWindows.of(Duration.of(2, MINUTES).toMillis())
  
      val newStream = stream.filter({ key, _ -&gt; isNew(key) })
              .map({key, value -&gt;  KeyValue(key.replace(NEW_PREFIX, ""), value) })
  
      val oldStream = stream.filter({ key, _ -&gt; isOld(key) })
              .map({key, value -&gt;  KeyValue(key.replace(OLD_PREFIX, ""), value) })
  
      val joined = newStream.join(oldStream,
              { value1, value2 -&gt; "$value1&amp;$value2" }, windows)
  
      joined.foreach({ key, value -&gt;
          log.info { "JOINED $key : $value" }
      })
  </code></pre>
  
  <p>Now I want to know new/old keys which are missing in <strong>time window</strong> for some reason. Is it possible to achieve with KafkaStreams API?</p>
  
  <p>In my case when key <code>"1-old"</code> is received and <code>"1-new"</code> is not within 2 minutes only in this case I want to report id <code>1</code> as suspicious. </p>