title: How to send record on topic when window is closed in kafka streams
body: <p>So i have been struggeling with this for a couple of days, acctually. I am consuming records from 4 topics. I need to aggregate the records over a TimedWindow. When the time is up, i want to send either an approved message or a not approved message to a sink topic. Is this possible to do with kafka streams?</p>
  
  <p>It seems it sinks every record to the new topic, even though the window is still open, and that's really not what i want.</p>
  
  <p>Here is the simple code:</p>
  
  <pre><code> builder.stream(getTopicList(), Consumed.with(Serdes.ByteArray(), 
   Serdes.ByteArray()))
  .flatMap(new ExceptionSafeKeyValueMapper&lt;String, 
   FooTriggerMessage&gt;("", Serdes.String(),
         fooTriggerSerde))
   .filter((key, value) -&gt; value.getTriggerEventId() != null)
   .groupBy((key, value) -&gt; value.getTriggerEventId().toString(),
         Serialized.with(Serdes.String(), fooTriggerSerde))
  
  .windowedBy(TimeWindows.of(TimeUnit.SECONDS.toMillis(30))
  .advanceBy(TimeUnit.SECONDS.toMillis(30)))
  
  .aggregate(() -&gt; new BarApprovalMessage(), /* initializer */
         (key, value, aggValue) -&gt; getApproval(key, value, aggValue),/*adder*/
         Materialized
                 .&lt;String, BarApprovalMessage, WindowStore&lt;Bytes, byte[]&gt;&gt;as(
                         storeName) /* state store name */
                 .withValueSerde(barApprovalSerde))
  .toStream().to(appProperties.getBarApprovalEngineOutgoing(), 
  Produced.with(windowedSerde, barApprovalSerde));
  </code></pre>
  
  <p>As of now, every record is being sinked to the outgoingTopic, i only want it to send one message when the window is closed, so to speak.</p>
  
  <p>Is this possible?</p>