@Override
  public void describe(StreamApplicationDescriptor appDescriptor) {
    KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");

    KafkaInputDescriptor<PageViewEvent> inputStreamDescriptor =
        trackingSystem.getInputDescriptor("pageViewEvent", new JsonSerdeV2<>(PageViewEvent.class));

    KafkaOutputDescriptor<KV<String, StatsOutput>> outputStreamDescriptor =
        trackingSystem.getOutputDescriptor("pageViewEventPerMember",
            KVSerde.of(new StringSerde(), new JsonSerdeV2<>(StatsOutput.class)));

    appDescriptor.withDefaultSystem(trackingSystem);
    MessageStream<PageViewEvent> pageViewEvents = appDescriptor.getInputStream(inputStreamDescriptor);
    OutputStream<KV<String, StatsOutput>> pageViewEventPerMember = appDescriptor.getOutputStream(outputStreamDescriptor);

    pageViewEvents
        .partitionBy(pve -> pve.getMemberId(), pve -> pve,
            KVSerde.of(new StringSerde(), new JsonSerdeV2<>(PageViewEvent.class)), "partitionBy")
        .map(KV::getValue)
        .flatMap(new MyStatsCounter())
        .map(stats -> KV.of(stats.memberId, stats))
        .sendTo(pageViewEventPerMember);
  }
Example code using {@link KeyValueStore} to implement event-time window
samza-test/src/main/java/org/apache/samza/example/KeyValueStoreExample.java