@Override
    public Collection<StatsOutput> apply(PageViewEvent message) {
      List<StatsOutput> outputStats = new ArrayList<>();
      long wndTimestamp = (long) Math.floor(TimeUnit.MILLISECONDS.toMinutes(message.getTimestamp()) / 5) * 5;
      String wndKey = String.format("%s-%d", message.getMemberId(), wndTimestamp);
      StatsWindowState curState = this.statsStore.get(wndKey);
      if (curState == null) {
        curState = new StatsWindowState();
      }
      curState.newCount++;
      long curTimeMs = System.currentTimeMillis();
      if (curState.newCount > 0 && curState.timeAtLastOutput + timeoutMs < curTimeMs) {
        curState.timeAtLastOutput = curTimeMs;
        curState.lastCount += curState.newCount;
        curState.newCount = 0;
        outputStats.add(new StatsOutput(message.getMemberId(), wndTimestamp, curState.lastCount));
      }
      // update counter w/o generating output
      this.statsStore.put(wndKey, curState);
      return outputStats;
    }
Example code using {@link KeyValueStore} to implement event-time window
samza-test/src/main/java/org/apache/samza/example/KeyValueStoreExample.java