@Override
  public void describe(StreamApplicationDescriptor appDescriptor) {
    JsonSerdeV2<PageView> inputSerde = new JsonSerdeV2<>(PageView.class);
    KVSerde<String, Integer> outputSerde = KVSerde.of(new StringSerde(), new IntegerSerde());
    KafkaSystemDescriptor ksd = new KafkaSystemDescriptor(SYSTEM);
    KafkaInputDescriptor<PageView> id = ksd.getInputDescriptor(INPUT_TOPIC, inputSerde);
    KafkaOutputDescriptor<KV<String, Integer>> od = ksd.getOutputDescriptor(OUTPUT_TOPIC, outputSerde);

    MessageStream<PageView> pageViews = appDescriptor.getInputStream(id);
    OutputStream<KV<String, Integer>> outputStream = appDescriptor.getOutputStream(od);

    pageViews
        .filter(m -> !FILTER_KEY.equals(m.getUserId()))
        .window(Windows.keyedTumblingWindow(PageView::getUserId, Duration.ofSeconds(3),
            new StringSerde(), new JsonSerdeV2<>(PageView.class)), "tumblingWindow")
        .map(m -> KV.of(m.getKey().getKey(), m.getMessage().size()))
        .sendTo(outputStream);

  }
A {@link StreamApplication} that demonstrates a filter followed by a tumbling window.
samza-test/src/test/java/org/apache/samza/test/operator/TumblingWindowApp.java