public static <M, K, WV> Window<M, K, WV> keyedTumblingWindow(MapFunction<? super M, ? extends K> keyFn, Duration interval,
      SupplierFunction<? extends WV> initialValue, FoldLeftFunction<? super M, WV> aggregator, Serde<K> keySerde,
      Serde<WV> windowValueSerde) {

    Trigger<M> defaultTrigger = new TimeTrigger<>(interval);
    return new WindowInternal<>(defaultTrigger, (SupplierFunction<WV>) initialValue, (FoldLeftFunction<M, WV>) aggregator,
        (MapFunction<M, K>) keyFn, null, WindowType.TUMBLING, keySerde, windowValueSerde, null);
  }
Creates a {@link Window} that groups incoming messages into fixed-size, non-overlapping processing
time based windows based on the provided keyFn and applies the provided fold function to them.

<p>The below example computes the maximum value per-key over fixed size 10 second windows.

<pre> {@code
   MessageStream<UserClick> stream = ...;
   MapFunction<UserClick, String> keyFn = ...;
   SupplierFunction<Integer> initialValue = () -> 0;
   FoldLeftFunction<UserClick, Integer, Integer> maxAggregator = (m, c) -> Math.max(parseInt(m), c);
   MessageStream<WindowPane<String, Integer>> windowedStream = stream.window(
       Windows.keyedTumblingWindow(keyFn, Duration.ofSeconds(10), maxAggregator));
}
</pre>

@param keyFn the function to extract the window key from a message
@param interval the duration in processing time
@param initialValue the initial value supplier for the aggregator. Invoked when a new window is created.
@param aggregator the function to incrementally update the window value. Invoked when a new message
                  arrives for the window.
@param keySerde the serde for the window key
@param windowValueSerde the serde for the window value
@param <M> the type of the input message
@param <WV> the type of the {@link WindowPane} output value
@param <K> the type of the key in the {@link Window}
@return the created {@link Window} function.
samza-api/src/main/java/org/apache/samza/operators/windows/Windows.java