title: How can I count users that generate events within a certain period of time with Kafka Streams?
body: <p>I have streaming events which has user-id in it. I want to count how many distinct user generate an event within certain of time. However, I am beginner in Kafka and I cannot cope with the problem.</p>
  <p>Example events in 1 minutes;</p>
  <pre><code>{&quot;event_name&quot;: &quot;viewProduct&quot;, &quot;user_id&quot;: &quot;12&quot;}
  {&quot;event_name&quot;: &quot;viewProductDetails&quot;, &quot;user_id&quot;: &quot;23&quot;}
  {&quot;event_name&quot;: &quot;viewProductComments&quot;, &quot;user_id&quot;: &quot;12&quot;}
  {&quot;event_name&quot;: &quot;viewProduct&quot;, &quot;user_id&quot;: &quot;23&quot;}
  {&quot;event_name&quot;: &quot;viewProductComments&quot;, &quot;user_id&quot;: &quot;32&quot;}
  </code></pre>
  <p>My code should generate there are 3 active users according to the events above.</p>
  <p>My approach is as follows, however this solution cannot eliminate multiple event from same user and count the same user multiple times.</p>
  <pre><code>builder.stream(&quot;orders&quot;) // read from orders toic
                  .mapValues(v -&gt; { // get user_id via json parser
                      JsonNode jsonNode = null;
                      try {
                          jsonNode = objectMapper.readTree((String) v);
                          return jsonNode.get(&quot;user_id&quot;).asText();
                      } catch (JsonProcessingException e) {
                          e.printStackTrace();
                      }
                          return &quot;&quot;;
                      })
                  .selectKey((k, v) -&gt; &quot;1&quot;) // put same key to every user_id
                  .groupByKey()
                  .windowedBy(TimeWindows.of(Duration.ofSeconds(1))) // use time windows
                  .count() // count values
  </code></pre>