@Override
    public void describe(StreamApplicationDescriptor appDescriptor) {
      DelegatingSystemDescriptor sd = new DelegatingSystemDescriptor(SYSTEM);
      GenericInputDescriptor<KV<String, PageView>> inputDescriptor =
          sd.getInputDescriptor(INPUT_TOPIC, KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()));
      GenericOutputDescriptor<KV<String, String>> outputDescriptor =
          sd.getOutputDescriptor(OUTPUT_TOPIC, KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()));

      appDescriptor.getInputStream(inputDescriptor)
          .map(KV::getValue)
          .partitionBy(PageView::getUserId, m -> m, KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()), "p1")
          .window(Windows.keyedSessionWindow(KV::getKey, Duration.ofSeconds(3), () -> 0, (m, c) -> c + 1, new StringSerde("UTF-8"), new IntegerSerde()), "w1")
          .map(wp -> KV.of(wp.getKey().getKey(), wp.getKey().getKey() + " " + wp.getMessage()))
          .sendTo(appDescriptor.getOutputStream(outputDescriptor));
    }
A {@link StreamApplication} that demonstrates a repartition followed by a windowed count.
samza-test/src/test/java/org/apache/samza/test/operator/TestRepartitionWindowApp.java