public static <M, K> Window<M, K, Collection<M>> keyedTumblingWindow(MapFunction<M, K> keyFn, Duration interval,
      Serde<K> keySerde, Serde<M> msgSerde) {

    Trigger<M> defaultTrigger = new TimeTrigger<>(interval);
    return new WindowInternal<>(defaultTrigger, null, null, keyFn, null, WindowType.TUMBLING,
        keySerde, null, msgSerde);
  }
Creates a {@link Window} that groups incoming messages into fixed-size, non-overlapping
processing time based windows using the provided keyFn.

<p>The below example groups the stream into fixed-size 10 second windows for each key.

<pre> {@code
   MessageStream<UserClick> stream = ...;
   Function<UserClick, String> keyFn = ...;
   MessageStream<WindowPane<String, Collection<UserClick>>> windowedStream = stream.window(
       Windows.keyedTumblingWindow(keyFn, Duration.ofSeconds(10)));
}
</pre>

@param keyFn function to extract key from the message
@param interval the duration in processing time
@param keySerde the serde for the window key
@param msgSerde the serde for the input message
@param <M> the type of the input message
@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