title: Kafka Streams | Aggregation of Joined stream
body: <p>I want to join two topic streams (left join) and do a window-based aggregation over the joined stream. However, the aggregation is counting some of the messages twice as during join some messages are emitted twice depending on the delay in the right topic. 
  Following is the code for the POC. </p>
  
  <pre><code>        StreamsBuilder builder = new StreamsBuilder();
          KStream&lt;Long, BidMessage&gt; bidStream = builder.stream("bid", Consumed.with(new LongSerde(), new BidMessageSerde()).withTimestampExtractor(new BidMessageTimestampExtractor()));
          KStream&lt;Long, ClickMessage&gt; clickStream = builder.stream("click", Consumed.with(new LongSerde(), new ClickMessageSerde()).withTimestampExtractor(new ClickMessageTimestampExtractor()));
  
          KStream&lt;String, BidMessage&gt; newBidStream = bidStream.selectKey((key, value) -&gt; value.getRequestId());
          KStream&lt;String, CLickMessage&gt; newClickStream = impStream.selectKey((key, value) -&gt; value.getRequestId());
  
          KStream&lt;String, BidMergedMessage&gt; result = newBidStream.leftJoin(newImpStream,
                  getValueJoiner(),
                  JoinWindows.of(Duration.ofSeconds(30)).grace(Duration.ofSeconds(0)),
                  Joined.with(Serdes.String(), new BidMessageSerde(), new ClickMessageSerde()));
  
          result.groupBy((key, value) -&gt; "" + value.getClientId(), Grouped.with(Serdes.String(), newBidMergedSerde()))
                  .windowedBy(TimeWindows.of(Duration.ofSeconds(30)).grace(Duration.ofSeconds(40)))
                  .aggregate(() -&gt; new AggResult(0, 0), (key, value, aggregate) -&gt; {
                      if (value.getClickId() != null) {
                          aggregate.clicks_++;
                      }
                      aggregate.bids_++;
                      return aggregate;
                  }, Materialized.with(Serdes.String(),new AggResultJsonSerde()))
                  .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
                  .toStream()
                  .foreach((key, value) -&gt; {
                      logger.info("{}-{}, clientId : {}, Value: {}", new Date(key.window().start()), new Date(key.window().end()),key.key(), value);
                  });
  
          final KafkaStreams streams = new KafkaStreams(builder.build(), props);
          streams.start();
  </code></pre>
  
  <p>Can it be fixed to avoid the duplicates due to join?</p>