public static <M> Window<M, Void, Collection<M>> tumblingWindow(Duration duration, Serde<M> msgSerde) {
    Trigger<M> defaultTrigger = new TimeTrigger<>(duration);
    return new WindowInternal<>(defaultTrigger, null, null, null, null,
        WindowType.TUMBLING, null, null, msgSerde);
  }
Creates a {@link Window} that groups incoming messages into fixed-size, non-overlapping
processing time based windows.

<p>The below example groups the stream into fixed-size 10 second windows and computes a windowed-percentile.

<pre> {@code
   MessageStream<Long> stream = ...;
   Function<Collection<Long>, Long> percentile99 = ..

   MessageStream<WindowPane<Void, Collection<Long>>> windowedStream =
       integerStream.window(Windows.tumblingWindow(Duration.ofSeconds(10)));
   MessageStream<Long> windowedPercentiles =
       windowedStream.map(windowPane -> percentile99(windowPane.getMessage());
}
</pre>

@param duration the duration in processing time
@param msgSerde the serde for the input message
@param <M> the type of the input message
@return the created {@link Window} function
samza-api/src/main/java/org/apache/samza/operators/windows/Windows.java