private long getWindowTimestamp(M message) {
    if (window.getWindowType() == WindowType.TUMBLING) {
      long triggerDurationMs = ((TimeTrigger<M>) window.getDefaultTrigger()).getDuration().toMillis();
      final long now = clock.currentTimeMillis();
      // assign timestamp to be the start timestamp of the window boundary
      long timestamp = now - now % triggerDurationMs;
      return timestamp;
    } else {
      K key = keyFn.apply(message);
      // get the value with the earliest timestamp for the provided key.
      ClosableIterator<TimestampedValue<Object>> iterator = timeSeriesStore.get(key, 0, Long.MAX_VALUE, 1);
      List<TimestampedValue<Object>> timestampedValues = toList(iterator);

      // If there are no existing sessions for the key, we return the current timestamp. If not, return the
      // timestamp of the earliest message.
      long timestamp = (timestampedValues.isEmpty())? clock.currentTimeMillis() : timestampedValues.get(0).getTimestamp();

      return timestamp;
    }
  }
Computes the timestamp of the window this message should belong to.

In the case of tumbling windows, timestamp of a window is defined as the start timestamp of its corresponding window
interval. For instance, if the tumbling interval is 10 seconds, all messages that arrive between [1000, 1010]
are assigned to the window with timestamp "1000"

In the case of session windows, timestamp is defined as the timestamp of the earliest message in the window.
For instance, if the session gap is 10 seconds, and the first message in the window arrives at "1002" seconds,
all messages (that arrive within 10 seconds of their previous message) are assigned a timestamp "1002".

@param message the input message
@return the timestamp of the window this message should belong to
samza-core/src/main/java/org/apache/samza/operators/impl/WindowOperatorImpl.java